Gelesen von Kafka und schreiben auf hdfs in Parkett

Zum Lesen von Daten aus Kafka und das schreiben auf HDFS in Parkett-format, Verwendung von Spark-Batch-job-statt-streaming, die Sie verwenden können,Spark Strukturierte Streaming.

Strukturierte Streaming ist eine skalierbare und fehlertolerante stream-processing-engine gebaut, auf dem Spark-SQL-engine. Sie können express Ihre streaming-Berechnung auf die gleiche Weise Ausdrücken würde eine batch-Berechnung auf statischen Daten. Die Spark SQL-engine kümmert es läuft inkrementell und kontinuierlich und aktualisieren das endgültige Ergebnis als streaming-Daten weiter zu kommen. Können Sie das Dataset/DataFrame API in Scala, Java, Python oder R, um die express-streaming-Aggregationen, event-time windows, stream-to-batch-joins, etc. Die Berechnung erfolgt auf der gleichen optimiert Spark SQL-engine. Schließlich gewährleistet das system die Ende-zu-Ende-exactly-once-Fehler-Toleranz gewährleistet durch checkpointing-und Write-Ahead-Logs. In kurzen, Strukturierten Streaming bietet schnelle, skalierbare, fehlertolerante, Ende-zu-Ende-exactly-once-stream-Verarbeitung, ohne dass der Benutzer Grund über streaming.

Kommt es mit Kafka als einen integrierten Source, d.h., wir können die Umfrage-Daten von Kafka. Es ist kompatibel mit Kafka-broker-Versionen 0.10.0 oder höher.

Für das ziehen der Daten von Kafka im batch-Modus, können Sie ein Dataset erstellen/DataFrame für einen definierten Bereich von offsets.

//Subscribe to 1 topic defaults to the earliest and latest offsets
val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

//Subscribe to multiple topics, specifying explicit Kafka offsets
val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribe", "topic1,topic2")
  .option("startingOffsets", """{"topic1":{"0":23,"1":-2},"topic2":{"0":-2}}""")
  .option("endingOffsets", """{"topic1":{"0":50,"1":-1},"topic2":{"0":-1}}""")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

//Subscribe to a pattern, at the earliest and latest offsets
val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
  .option("subscribePattern", "topic.*")
  .option("startingOffsets", "earliest")
  .option("endingOffsets", "latest")
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

Jede Zeile in der Quelle hat das folgende schema:

| Column           | Type          |
|:----------------- | --------------:|
| key              |        binary |
| value            |        binary |
| topic            |        string |
| partition        |           int |
| offset           |          long |
| timestamp        |          long |
| timestampType    |           int |

Nun, zum schreiben von Daten in HDFS in Parkett-format, folgenden code geschrieben werden kann:

df.write.parquet("hdfs://data.parquet")

Weitere Informationen zu Funke Strukturierte Streaming - + Kafka, entnehmen Sie bitte folgenden Leitfaden - Kafka Integration Guide

Ich hoffe, es hilft!

Verwenden Kafka-Streams. SparkStreaming ist ein Etikettenschwindel (es ist mini-batch unter der Haube, zumindest bis zu 2,2).

https://eng.verizondigitalmedia.com/2017/04/28/Kafka-to-Hdfs-ParquetSerializer/