无法使用 spark 过滤字符串结构答案

【问题标题】：Cannot filter a strucure of Strings with spark无法使用 spark 过滤字符串结构
【发布时间】：2021-03-23 21:58:19
【问题描述】：

我正在尝试从具有这种结构的数据框中过滤行：

 |-- age: integer (nullable = true)
 |-- qty: integer (nullable = true)
 |-- dates: array (nullable = true)
 |    |-- element: timestamp (containsNull = true)

例如，在这个数据框中，我只想要第一行：

+---------+------------+------------------------------------------------------------------+
|    age  | qty        |dates                                                             |
+---------+------------+------------------------------------------------------------------+
|  54     |           1|  [2020-12-31 12:15:20, 2021-12-31 12:15:20]                      |
|  45     |           1|  [2020-12-31 12:15:20, 2018-12-31 12:15:20, 2019-12-31 12:15:20] |
+---------+------------+------------------------------------------------------------------+

这是我的代码：

 val result = sqlContext
     .table("scores")
 
 
 result.filter(array_contains(col("dates").cast("string"),
 2021)).show(false)

但是我收到了这个错误：

org.apache.spark.sql.AnalysisException: 无法解析 'array_contains( 由于数据类型不匹配：参数必须是一个数组，后跟一个与 > 数组成员相同类型的值；

有人可以帮忙吗？

【问题讨论】：

标签： scala apache-spark apache-spark-sql

【解决方案1】：

您需要使用rlike 来检查每个数组元素是否包含2021。array_contains 检查完全匹配，而不是部分匹配。

result.filter("array_max(transform(dates, x -> string(x) rlike '2021'))").show(false)

【讨论】：

【解决方案2】：

您可以分解 ArrayType，然后根据需要进行处理：将列转换为字符串，然后应用您的过滤器：

 val spark: SparkSession = SparkSession.builder()
      .master("local[*]")
      .appName("SparkByExamples")
      .getOrCreate()

    spark.sparkContext.setLogLevel("ERROR")

    import java.sql.Timestamp
    import java.text.SimpleDateFormat
    def convertToTimeStamp(s: String) = {
      val dateFormat = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss")
      val parsedDate = dateFormat.parse(s)
      new Timestamp(parsedDate.getTime)

    }


    val data = Seq(
      Row(54, 1, Array(convertToTimeStamp("2020-12-31 12:15:20"), convertToTimeStamp("2021-12-31 12:15:20"))),
      Row(45, 1, Array(convertToTimeStamp("2020-12-31 12:15:20"), convertToTimeStamp("2018-12-31 12:15:20"), convertToTimeStamp("2019-12-31 12:15:20")))
    )
    val Schema = StructType(Array(
      StructField("age", IntegerType, nullable = true),
      StructField("qty", IntegerType, nullable = true),
      StructField("dates", ArrayType(TimestampType, containsNull = true), nullable = true)

    ))

    val rdd = spark.sparkContext.parallelize(data)
    var df = spark.createDataFrame(rdd, Schema)
    df.show()
    df.printSchema()
    df = df.withColumn("exp",f.explode(f.col("dates")))
    df.filter(f.col("exp").cast(StringType).contains("2021")).show()

【讨论】：

【解决方案3】：

您可以使用exists 函数来检查dates 数组是否包含2021 年的日期：

df.filter("exists(dates, x -> year(x) = 2021)").show(false)

//+---+---+------------------------------------------+
//|age|qty|dates                                     |
//+---+---+------------------------------------------+
//|54 |1  |[2020-12-31 12:15:20, 2021-12-31 12:15:20]|
//+---+---+------------------------------------------+

如果要使用array_contains，则需要将时间戳元素转换为年份：

df.filter("array_contains(transform(dates, x -> year(x)), 2021)").show(false)

【讨论】：