Modulul Spark Structured Streaming tratează fluxurile de date continue ca pe un tabel infinit la care se adaugă neîncetat rânduri noi. Prin conectorul oficial Kafka, Spark consumă mesaje din topic-uri cu gestionarea automată a offset-urilor și toleranță completă la defecte (Exactly-Once processing).
Exemplu: Conectarea la Kafka și extragerea payload-ului JSON
from pyspark.sql import functions as F
df_stream_kafka = spark.readStream \
.format("kafka") \
.option("kafka.bootstrap.servers", "kafka.dexblog.ro:9092") \
.option("subscribe", "evenimente-plati") \
.option("startingOffsets", "latest") \
.load()
# Conversia cheii și a valorii din octeți binari în text String:
df_evenimente = df_stream_kafka.selectExpr(
"CAST(key AS STRING) as id_tranzactie",
"CAST(value AS STRING) as payload_json",
"timestamp"
)
Întrebări & Exemple de Cod:
Ingestia în Timp Real din Kafka cu readStreamAi o nelămurire despre utilizarea comenzii
Ingestia în Timp Real din Kafka cu readStreamsau ai descoperit un truc / fragment de cod util? Distribuie-l cu ceilalți programatori!Fii primul care pune o întrebare sau distribuie un fragment de cod / prompt!
Nu au fost adăugate încă întrebări sau exemple de cod pentru această pagină. Dacă ai întâlnit o problemă sau ai un snippet util, adaugă-l mai jos pentru a-i ajuta pe ceilalți.