Javaで構造化ストリーミングを使用してKafkaからレコードを逆シリアル化する方法は？

Question

私はスパーク2.1を使用します。

私はSpark Structured Streamingを使用してKafkaからレコードを読み込み、逆シリアル化してから集計を適用しようとしています。

私は次のコードを持っている：私が欲しいもの

SparkSession spark = SparkSession 
      .builder() 
      .appName("Statistics") 
      .getOrCreate(); 

    Dataset<Row> df = spark 
      .readStream() 
      .format("kafka") 
      .option("kafka.bootstrap.servers", kafkaUri) 
      .option("subscribe", "Statistics") 
      .option("startingOffsets", "earliest") 
      .load(); 

    df.selectExpr("CAST(value AS STRING)") 

ではなくStringとしてキャストする私のオブジェクトにvalueフィールドをデシリアライズすることです。

私はこれのためのカスタムデシリアライザを持っています。

public StatisticsRecord deserialize(String s, byte[] bytes) 

Javaでこれを行うにはどうすればよいですか？

---

私が見つけた唯一の関連リンクはhttps://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.htmlですが、これはScala向けです。

Answer 1

JSONメッセージのスキーマを定義します。

StructType schema = DataTypes.createStructType(new StructField[] { 
       DataTypes.createStructField("Id", DataTypes.IntegerType, false), 
       DataTypes.createStructField("Name", DataTypes.StringType, false), 
       DataTypes.createStructField("DOB", DataTypes.DateType, false) }); 

今読むメッセージは次のとおりです。 MessageDataはJSONメッセージのJavaBeanです。あなたのデータのためのJavaのカスタムデシリアライザを持っている場合は

Dataset<MessageData> df = spark 
      .readStream() 
      .format("kafka") 
      .option("kafka.bootstrap.servers", kafkaUri) 
      .option("subscribe", "Statistics") 
      .option("startingOffsets", "earliest") 
      .load() 
      .selectExpr("CAST(value AS STRING) as message") 
      .select(functions.from_json(functions.col("message"),schema).as("json")) 
      .select("json.*") 
      .as(Encoders.bean(MessageData.class)); 

Answer 2

は、あなたがload後カフカから入手バイトにそれを使用しています。

df.select("value") 

そのラインはあなただけの単一の列valueとDataset<Row>を与えます。

---

私はスカラ座のSpark APIを排他的に私は、「直列化復元」の場合処理するために、Scalaで次の操作を実行したいです：

import org.apache.spark.sql.Encoders 
implicit val statisticsRecordEncoder = Encoders.product[StatisticsRecord] 
val myDeserializerUDF = udf { bytes => deserialize("hello", bytes) } 
df.select(myDeserializerUDF($"value") as "value_des") 

何をしたいあなたを与える必要があります... Scalaで

カスタムオブジェクトにエンコーダを使用可能にする必要があること、またはSpark SQLはそのオブジェクトをデータセット内に置くことを拒否します。