java - 运行时在ApacheIndex Aggregate函数中运行到ArrayIndexOutOfBoundsException

在将Java Spark程序部署到群集期间，它遇到ArrayIndexOutOfBoundsException: 11异常

根据我的理解，我的代码的编写在语法上没有什么错，并且索引错误并不能说明问题所在。我的程序只是应该能够容纳12列，并用空格隔开。从那时起，它需要接受ONE列（命令列）并进行汇总以查看每个命令存在多少次，即

column1 column2 command column3 ect
dggd     gdegdg  cmd#1   533    ect
dggd     gdegdg  cmd#1   533    ect
dggd     gdegdg  cmd#2   534    ect
dggd     gdegdg  cmd#5   5353   ect
dggd     gdegdg  cmd#2   533    ect

看起来像

commmand    count
command#1    5
command#2    15
command#5    514

我在跑
火花2.1
HDP 2.6
这是我到目前为止的代码

public class Main {
public static void main(String[] args) {
    //functions fu = new functions();

 JavaSparkContext sc = new JavaSparkContext(new SparkConf().setAppName("appName").setMaster("local[*]"));
 SparkSession spark = SparkSession
            .builder()
            .appName("Log File Reader")
            .getOrCreate();

    JavaRDD<String> logsRDD = spark.sparkContext()
            .textFile(args[0],1)
            .toJavaRDD();

    String schemaString = "date time command service responseCode bytes ip dash1 dash2 dash3 num dash4";

    List<StructField> fields = new ArrayList<>();
    String[] fieldName = schemaString.split(" ");

    for (String field : fieldName){
        fields.add(DataTypes.createStructField(field, DataTypes.StringType, true));
    }

    StructType schema = DataTypes.createStructType(fields);

    JavaRDD<Row> rowRDD = logsRDD.map((Function<String, Row>) record -> {
       String[] attributes = record.split(" ");
       return RowFactory.create(attributes[0],attributes[1],attributes[2],attributes[3],attributes[4],attributes[5],
               attributes[6],attributes[7],attributes[8],attributes[9],
               attributes[10],attributes[11]);
    });

    Dataset<Row> dataDF = spark.createDataFrame(rowRDD, schema);

    dataDF.createOrReplaceTempView("data");

    //shows the top 20 rows from the dataframe including all columns
    Dataset<Row> showDF = spark.sql("select * from data");

    //shows the top 20 columns from the same dataframe, but only displays
    //the command column
    Dataset<Row> commandDF = spark.sql("select command from data");
    showDF.show();
    commandDF.show();

这段代码可以正常工作，但是当我尝试使用如下代码查找最终结果时，就会遇到索引错误。

logsDF.groupBy(col("command")).count().show();
Dataset<Row> ans = spark.sql("select command, count(*) from logs group by command").show();

最后是提交火花的代码

spark-submit --class com.ect.java.Main /path/application.jar hdfs:///path/textfile.txt

我认为这是环境问题，但找不到与此相关的任何文档

最佳答案

问题不在于聚合函数。问题出在您的日志文件上。您遇到错误

Caused by: java.lang.ArrayIndexOutOfBoundsException: 11

这意味着您的日志文件中的一行中有11个条目，而不是12个，这是程序所要求的。您可以通过创建示例log.txt文件并在该文件中保留两行来验证这一点。

您的分组代码应如下所示（看起来像错字）。在示例应用程序中，您有dataDF而不是logsDF。临时表名称是data而不是logs。

dataDF.groupBy(col("command")).count().show();
Dataset<Row> ans = spark.sql("select command, count(*) from data group by command");
ans.show();