主页 / coding / 问题 / 79309612
Accepted
Ged
Asked: 2024-12-26 21:38:52 +0800 CST

Databricks 社区版中多个接收器处理无法持久化

  • 772

我只是想将 Rate 与 Structured Streaming 结合使用，以便将每个 MicroBatch 写入多个表名。即在 pyspark 中刷新多个接收器逻辑以准备某些认证。

没有错误，但没有发生持久性。我看的时候已经有一段时间了；一定是一些基本的东西。

在 Databricks 社区版上进行如下编码，没有 Hive 目录。基本内容。

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat, lit

spark = SparkSession.builder \
    .appName("SimulateKAFKAandMultipleSinks") \
    .getOrCreate()

rate_stream = spark.readStream \
    .format("rate") \
    .option("rowsPerSecond", 1) \
    .load()

message_stream = rate_stream.select(
    (rate_stream["value"] + 1000).alias("message_id"),              
    rate_stream["timestamp"].alias("event_time"),                    
    (concat(lit("T"), rate_stream["value"] % 5)).alias("table_id")   
)

def append_to_parquet(df, table_id):
    table_path = f"/mnt/parquet/{table_id}"   

    df.write \
        .format("parquet") \
        .mode("append") \
        .option("path", table_path) \
        .save()

def process_batch(df, batch_id):
    partitioned_df = df.repartition("table_id")
    def process_partition(iterator):
        for partition in iterator:
            first_row_value = df.first()
            table_id_value = first_row_value['table_id']
            print(f"Writing partition for table_id: {table_id_value}") 
            partition_df = partition.filter(col("table_id") == table_id_value)
            append_to_parquet(partition_df, table_id_value)

    partitioned_df.rdd.mapPartitions(process_partition)

query = message_stream.writeStream \
    .foreachBatch(process_batch) \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/parquet/checkpoints/") \
    .start()

query.awaitTermination()

更新。我这边有疏忽，但足够有教育意义。可以使用批处理 KAFKA 来实现，而不是通过结构化流。

apache-spark

1 个回答

  1. Best Answer

    您的代码无法持久保存数据的问题可能源于您在 Spark 中使用结构化流的几个基本问​​题。

    试试这个

    from pyspark.sql import SparkSession
from pyspark.sql.functions import col, concat, lit

# Create Spark session
spark = SparkSession.builder \
    .appName("SimulateKAFKAandMultipleSinks") \
    .getOrCreate()

# Create a streaming DataFrame
rate_stream = spark.readStream \
    .format("rate") \
    .option("rowsPerSecond", 1) \
    .load()

# Transform the data
message_stream = rate_stream.select(
    (rate_stream["value"] + 1000).alias("message_id"),
    rate_stream["timestamp"].alias("event_time"),
    concat(lit("T"), (rate_stream["value"] % 5).cast("string")).alias("table_id")
)

# Write to multiple sinks based on table_id
def write_to_multiple_sinks(batch_df, batch_id):
    unique_table_ids = batch_df.select("table_id").distinct().collect()
    for row in unique_table_ids:
        table_id = row["table_id"]
        table_path = f"/mnt/parquet/{table_id}"  # Replace with actual storage path
        table_df = batch_df.filter(col("table_id") == table_id)
        table_df.write.mode("append").parquet(table_path)

query = message_stream.writeStream \
    .foreachBatch(write_to_multiple_sinks) \
    .outputMode("append") \
    .option("checkpointLocation", "/mnt/parquet/checkpoints/") \
    .start()

query.awaitTermination()
    • 1

相关问题