взять образец формы потокового фрейма данных

Question

Я пытаюсь применить функцию (которая работает с обычными кадрами данных) для потоковой передачи данных. Перед применением этой функции мне нужно использовать .rdd.takeSample () для заданных данных, но, конечно, это не работает с потоковыми фреймами данных.

Я получаю свои потоковые данные, используя следующий структурированный код потоковой передачи:

dsraw = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", "192.168.99.100:9092") \
            .option("subscribe", "topic") \
            .option("startingOffsets", "earliest") \
            .load()

ds = dsraw.selectExpr("CAST(value AS STRING)")

Мои данные - это набор случайных чисел в виде {'число': 1} и т. Д. c. В идеале я хочу поместить все числа, считанные из этого потока, в фрейм данных и вернуть его.

Есть ли способ конвертировать потоковый фрейм данных в искровой фрейм данных или в rdd? Если нет, есть ли альтернативный метод для takeSample?

Answer 1

Один из способов сделать это - записать потоковые данные в память,
, а затем создать фрейм данных / rdd, используя spark sql:

dsraw = spark \
            .readStream \
            .format("kafka") \
            .option("kafka.bootstrap.servers", "192.168.99.100:9092") \
            .option("subscribe", "topic") \
            .option("startingOffsets", "earliest") \
            .load()

ds = dsraw.selectExpr("CAST(value AS STRING)")

kafka_value_df = ds.selectExpr("CAST(value AS STRING)")
output_query = kafka_value_df.writeStream \
                      .queryName("numbers") \
                      .format("memory") \
                      .start()
output_query.awaitTermination(10)

value_df = spark.sql("select * from numbers")  # df

value_rdd = value_df.rdd  # rdd

Я не знаю точно, в каком формате ваш оригинал данные (просто {'number': 1} недостаточно), вам может потребоваться использовать map или json.loads в зависимости от ваших данных, чтобы получить нужный формат для df / rdd.