在 Spark Structured Streaming 中指定“basePath”选项答案

【问题标题】：Specifying "basePath" option in Spark Structured Streaming在 Spark Structured Streaming 中指定“basePath”选项
【发布时间】：2018-08-09 00:37:18
【问题描述】：

在 Spark Structured Streaming（Java 中）中读取分区数据时是否可以设置 basePath 选项？我只想加载特定分区中的数据，例如basepath/x=1/，但我也希望将x 作为列加载。设置basePath 为非流式数据帧的方式似乎不起作用。

这是一个最小的例子。我有一个包含以下数据的数据框：

+---+---+
|  a|  b|
+---+---+
|  1|  2|
|  3|  4|
+---+---+

我将此作为 Parquet 文件写入名为 x=1 的子目录中。

以下代码（使用常规的非流式数据帧）可以正常工作：

Dataset<Row> data = sparkSession.read()
  .option("basePath", basePath)
  .parquet(basePath + "/x=1");

data.show();

这会产生预期的结果：

+---+---+---+
|  a|  b|  x|
+---+---+---+
|  1|  2|  1|
|  3|  4|  1|
+---+---+---+

但是，以下（使用结构化流式处理 API）不起作用：

StructType schema = data.schema(); // data as defined above

Dataset<Row> streamingData = sparkSession.readStream()
  .schema(schema)
  .option("basePath", basePath)
  .parquet(basePath + "/x=1");

streamingData.writeStream()
  .trigger(Trigger.Once())
  .format("console")
  .start().awaitTermination();

在这种情况下，数据框不包含任何行：

+---+---+---+
|  a|  b|  x|
+---+---+---+
+---+---+---+

【问题讨论】：

是的，它不读取分区下的任何文件。

标签： java apache-spark spark-streaming

【解决方案1】：

我不确定这是否适用于火花流，但它适用于我在 Scala 中的批处理。我要做的是完全避免使用basePath。例如，当我的数据按年/月/日进行分区时，我想每天循环和处理，我会使用字符串插值。

import java.text.SimpleDateFormat
import java.sql.Timestamp
import java.util.Calendar

var dateStart: String = "01/14/2012"
var dateStop: String = "01/18/2012"

var  format: SimpleDateFormat = new SimpleDateFormat("MM/dd/yyyy");


var d1 = new Timestamp(format.parse(dateStart).getTime());
var d2 = new Timestamp(format.parse(dateStop).getTime());

var diffDays:Long = (d2.getTime() - d1.getTime()) / (24 * 60 * 60 * 1000)

var cal:Calendar = Calendar.getInstance()
cal.setTimeInMillis(d1.getTime())
for (i <- 0 to diffDays.toInt){
    val year = cal.get(Calendar.YEAR)
    val month = cal.get(Calendar.MONTH)
    val day = cal.get(Calendar.DAY_OF_MONTH)
    var dataframe1 = spark.read
           .load(s"s3://bucketName/somepath/year=$year/month=$month/day=$day")
    /*
    Do your dataframe manipulation here
    */
    cal.add(Calendar.DAY_OF_YEAR, 1)
}

您也可以使用字符串或整数列表来执行此操作。如果您需要将该数据视为一列，您始终可以将其作为新列附加到数据框中。不过，我不确定这是否适用于您使用 Spark 流的情况。

【讨论】：