使用Apache Spark从数据框中获取独特的计数

Question

我的数据看起来像这样

+--------------+---------+-------+---------+
|       dataOne|OtherData|dataTwo|dataThree|
+--------------+---------|-------+---------+
|          Best|     tree|      5|      533|
|            OK|     bush|      e|     3535|
|           MEH|      cow|      -|     3353|
|           MEH|      oak|   none|       12|
+--------------+---------+-------+---------+

而我正试图把它输入到输出中

+--------------+---------+
|       dataOne|    Count|
+--------------+---------|
|          Best|        1|
|            OK|        1|
|           Meh|        2|
+--------------+---------+

我没有问题将dataOne单独放入数据框并显示它的内容以确保我只是抓住dataOne列，但是我似乎无法找到正确的语法来将sql查询转换为我需要的数据。我尝试从整个数据集创建的临时视图中创建以下数据帧

Dataset<Row> dataOneCount = spark.sql("select dataOne, count(*) from 
dataFrame group by dataOne");
dataOneCount.show();

但是火花我能够在此找到的文档仅显示了如何在spark 1.6和之前进行此类聚合，因此任何帮助都将受到赞赏。

这是我得到的错误消息，但是我检查了数据并且没有索引错误。

 java.lang.ArrayIndexOutOfBoundsException: 11

我也尝试应用函数（）方法countDistinct

Column countNum = countDistinct(dataFrame.col("dataOne"));
Dataset<Row> result = dataOneDataFrame.withColumn("count",countNum);
result.show();

其中dataOneDataFrame是从运行中创建的dataFrame

select dataOne from dataFrame

但是它返回一个分析异常，我仍然是新兴的，因此我不确定我是否/何时评估countDistinct方法时出现错误

编辑：为了澄清，显示的第一个表是我通过读取文本文件并为其应用自定义模式而创建的dataFrame的结果（它们仍然是所有字符串）

Dataset<Row> dataFrame 

这是我的完整代码

public static void main(String[] args) {


    SparkSession spark = SparkSession
            .builder()
            .appName("Log File Reader")
            .getOrCreate();

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

    String schemaString = "dataOne OtherData dataTwo dataThree";

    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]);
    });


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

    //first attempt
    dF.groupBy(col("dataOne")).count().show();

    //Trying with a sql statement
    dF.createOrReplaceTempView("view");
    dF.sparkSession().sql("select command, count(*) from view group by command").show();

想到的最可能的事情是使用RowFactory返回行的lambda函数？这个想法听起来似乎很合理，但我不确定它是如何真正起作用的，或者我是否有另一种方法可以做到这一点。除此之外，我很困惑

样本数据

best tree 5 533
OK bush e 3535
MEH cow - 3353
MEH oak none 12

Answer 1

使用Scala语法以方便使用。它与Java语法非常相似：

// Input data
val df = {
  import org.apache.spark.sql._
  import org.apache.spark.sql.types._
  import scala.collection.JavaConverters._

  val simpleSchema = StructType(
    StructField("dataOne", StringType) ::
    StructField("OtherData", StringType) ::
    StructField("dataTwo", StringType) ::
    StructField("dataThree", IntegerType) :: Nil)

  val data = List(
    Row("Best", "tree", "5", 533),
    Row("OK", "bush", "e", 3535),
    Row("MEH", "cow", "-", 3353),
    Row("MEH", "oak", "none", 12)
  )

  spark.createDataFrame(data.asJava, simpleSchema)
}

df.show

+-------+---------+-------+---------+
|dataOne|OtherData|dataTwo|dataThree|
+-------+---------+-------+---------+
|   Best|     tree|      5|      533|
|     OK|     bush|      e|     3535|
|    MEH|      cow|      -|     3353|
|    MEH|      oak|   none|       12|
+-------+---------+-------+---------+

df.groupBy(col("dataOne")).count().show()

+-------+-----+
|dataOne|count|
+-------+-----+
|    MEH|    2|
|   Best|    1|
|     OK|    1|
+-------+-----+

我可以使用S3上的四行数据文件提交上面给出的Java代码，它可以正常工作：

$SPARK_HOME/bin/spark-submit \
  --class sparktest.FromStackOverflow \
  --packages "org.apache.hadoop:hadoop-aws:2.7.3" \
  target/scala-2.11/sparktest_2.11-1.0.0-SNAPSHOT.jar "s3a://my-bucket-name/sample.txt"