Sparkのバッチデータセットにストリーミングデータセットを追加する

Question

データベースの履歴データをSparkにロードし、新しいストリーミングデータをSparkに追加し続けるSparkのユースケースを用意しています。最新のデータセット。

私が知る限り、Spark SQLもSpark Streamingも、履歴データとストリーミングデータを組み合わせることはできません。それから、Spark 2.0のストラクチャードストリーミングがこの問題のために構築されているようです。しかし、いくつかの実験の後、私はまだそれを理解することはできません。ここに私のコードは次のとおりです。

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(); 

私はエラーだ「org.apache.spark.sql.AnalysisException：/データセットがサポートされていないストリーミングやバッチデータフレームの間の連合を。」私は2つのデータセットを結合します。

誰でもお手伝いできますか？間違った方向に行くのですか？

Answer 1

このタイプの機能をサポートする点でMongoDBのスパークコネクタは話せません.Googleにはそのような機能はあまりないようです。しかし、スパークデータベースエコシステムには他にもデータベースがあります。私はanother answerでスパークデータベースエコシステムにあるもののほとんどをカバーしました。私はSnappyDataとMemSQLがそのリストにあることを知っていますが、探している機能の種類をどのデータベースが簡単に許可しているのかを正確に言うことはできません。しかし、両方のために、リレーショナル形式のデータが必要な場合があります。