我们在Spark中有一个使用案例,需要将历史数据从数据库加载到Spark,并继续添加新的流式数据到Spark,然后我们可以对整个最新数据集进行分析。
据我所知,无论是Spark SQL还是Spark Streaming都无法将历史数据和流式数据相结合。然后我发现了Spark 2.0中的Structured Streaming似乎就是为解决这个问题而构建的。但经过一些实验,我仍然无法搞清楚。以下是我的代码:
据我所知,无论是Spark SQL还是Spark Streaming都无法将历史数据和流式数据相结合。然后我发现了Spark 2.0中的Structured Streaming似乎就是为解决这个问题而构建的。但经过一些实验,我仍然无法搞清楚。以下是我的代码:
SparkSession spark = SparkSession
.builder()
.config(conf)
.getOrCreate();
JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
// Load historical data from MongoDB
JavaMongoRDD<Document> mongordd = MongoSpark.load(jsc);
// Create typed dataset with customized schema
JavaRDD<JavaRecordForSingleTick> rdd = mongordd.flatMap(new FlatMapFunction<Document, JavaRecordForSingleTick>() {...});
Dataset<Row> df = spark.sqlContext().createDataFrame(rdd, JavaRecordForSingleTick.class);
Dataset<JavaRecordForSingleTick> df1 = df.as(ExpressionEncoder.javaBean(JavaRecordForSingleTick.class));
// ds listens to a streaming data source
Dataset<Row> ds = spark.readStream()
.format("socket")
.option("host", "127.0.0.1")
.option("port", 11111)
.load();
// Create the typed dataset with customized schema
Dataset<JavaRecordForSingleTick> ds1 = ds
.as(Encoders.STRING())
.flatMap(new FlatMapFunction<String, JavaRecordForSingleTick>() {
@Override
public Iterator<JavaRecordForSingleTick> call(String str) throws Exception {
...
}
}, ExpressionEncoder.javaBean(JavaRecordForSingleTick.class));
// ds1 and df1 have the same schema. ds1 gets data from the streaming data source, df1 is the dataset with historical data
ds1 = ds1.union(df1);
StreamingQuery query = ds1.writeStream().format("console").start();
query.awaitTermination();
当我将两个数据集union()起来时,出现了错误信息"org.apache.spark.sql.AnalysisException: Union between streaming and batch DataFrames/Datasets is not supported;"。请问有人能帮忙解决吗?是我方向错了吗?
mapWithState
不是显而易见的选择,但是如果你不需要历史数据,为什么要在流中保留它呢? - Glennie Helles Sindholt