Потоковая передача Spark + интеграция с Kafka, чтение данных из Kafka каждые 15 минут и сохранение смещения последнего чтения с помощью PySpark

Используя PySpark, реализована потоковая передача искр + интеграция Kafka. При каждом запуске он дает смещение от 0.

Нужно решить 2 вопроса:

Чтение ленты за последние 15 минут
Чтение последнего зафиксированного смещения для каждого раздела.

Помогите в решении этого

# create spark session 
spark = SparkSession.builder \
    .appName(appName) \
    .config("spark.jars.packages", "org.apache.spark:spark-sql-kafka-0-10_2.12:3.5.1") \
    .getOrCreate()

# Define schema for data in value field
schema =  StructType([
            StructField("col1", StringType()),
            StructField("col2", StringType()),
            StructField("col3" , TimestampType()), 
            StructField("col4" , DoubleType())
            ])

# Spark streaming 
df = spark \
    .read \
    .format("kafka") \
    .option("kafka.bootstrap.servers", broker) \
    .option("subscribe", topic) \
    .option("kafka.group.id", appName) \
    .option("enable.auto.commit", True) \
    .load()


value_df = df.select(col("topic"), col("partition"), col("offset"), from_json(col("value").cast("STRING"), schema).alias("values"))

Проверьте этот ответ

— 22.05.2024 06:44

это было полезно, помогло решить проблему. Спасибо

— 22.05.2024 13:02

apache-spark pyspark apache-kafka apache-spark-sql offset

21.05.2024 13:14

Стоит ли изучать PHP в 2026-2027 годах?

Привет всем, сегодня я хочу высказать свои соображения по поводу вопроса, который я уже много раз получал в своем сообществе: "Стоит ли изучать PHP в...

Поведение ключевого слова "this" в стрелочной функции в сравнении с нормальной функцией

В JavaScript одним из самых запутанных понятий является поведение ключевого слова "this" в стрелочной и обычной функциях.

Приемы CSS-макетирования - floats и Flexbox

Здравствуйте, друзья-студенты! Готовы совершенствовать свои навыки веб-дизайна? Сегодня в нашем путешествии мы рассмотрим приемы CSS-верстки - в...

Тестирование функциональных ngrx-эффектов в Angular 16 с помощью Jest

В системе управления состояниями ngrx, совместимой с Angular 16, появились функциональные эффекты. Это здорово и делает код определенно легче для...

Концепция локализации и ее применение в приложениях React ⚡️

Локализация - это процесс адаптации приложения к различным языкам и культурным требованиям. Это позволяет пользователям получить опыт, соответствующий...

Пользовательский скаляр GraphQL

Листовые узлы системы типов GraphQL называются скалярами. Достигнув скалярного типа, невозможно спуститься дальше по иерархии типов. Скалярный тип...

Перейти к ответу Данный вопрос помечен как решенный

Ответы 1

Ответ принят как подходящий

См. Руководство по программированию структурированной потоковой передачи и Документацию по структурированной потоковой передаче Kafka.

spark.read() — пакетная обработка. spark.readStream() — структурированная потоковая передача.

Чтение стрима за последние 15 минут
Чтение последнего зафиксированного смещения для каждого раздела.

Одна основная вещь, которую нужно понять, это то, что (как и в Kafka) обе эти вещи (контрольная точка и частота) указаны для каждого потребителя. Итак, вы указываете это, когда writeStream, а не когда readStream.

Думайте о «потоковой передаче Dataframe, возвращаемой spark.readStream()» как о теме Кафки. Который, в свою очередь, может иметь несколько потребителей, и каждый потребитель может определять свою собственную контрольную точку, частоту и т. д.

Аналогичным образом вы можете иметь несколько потребителей для одного потокового Dataframe и указать для каждого разные контрольные точки и частоту. Например.

# Create streaming dataframe
sdf = spark.readStream.format('kafka').option(...)

# start() DataStreamWriter to get create streaming queries
sq1 = sdf.writeStream.parquet(...) \
  .trigger(processingTime='1 seconds').option('checkpointLocation', 'loc1').start()
sq2 = sdf.writeStream.csv(...) \
  .trigger(processingTime='5 seconds').option('checkpointLocation', 'loc2').start()

spark.streams.awaitTermination()

Чтение стрима за последние 15 минут

Это контролируется с помощью DataStreamWriter.trigger()

Этого можно добиться двумя способами:

Запускайте задание каждые X минут, оно будет читать все новые сообщения с момента последней контрольной точки. В вашем случае X=15, но код будет универсальным, т. е. вы можете запускать один и тот же код с любой частотой, не меняя его. Ключевое здесь .trigger(availableNow=True). Таким образом, ваш компьютер может запускаться каждые 15 минут, запускать задание и затем выключаться. Кластер блоков данных, например. запуск занимает 4-6 минут.
Пусть работа выполняется непрерывно. И он ищет новые сообщения каждые 15 минут. Таким образом, ваш компьютер работает все время. Для этого вы будете использовать .trigger(processingTime='15 minutes'). Если вам нужны более низкие задержки и/или постоянно работающий общий кластер, вы можете сделать это, в противном случае нет смысла делать это с частотой 15 минут.