Writing unique parquet file per windows with Apache Beam Python

Question

I am trying to stream messages from kafka consumer to google cloud storage with 30 seconds windows using apache beam. Used beam_nuggets.io for reading from a kafka topic. However, I wasn't able to write unique parquet files to GCS per each window. You can see my code below:`

import apache_beam as beam
from apache_beam.transforms.trigger import AfterAny, AfterCount, AfterProcessingTime, AfterWatermark, Repeatedly

from apache_beam.portability.api.beam_runner_api_pb2 import AccumulationMode
from apache_beam.options.pipeline_options import PipelineOptions
from beam_nuggets.io import kafkaio
import json
from datetime import datetime
import pandas as pd
import config as conf
import apache_beam.transforms.window as window
consumer_config = {"topic": "Uswrite",
                   "bootstrap_servers": "*.*.*.*:9092",
                   "group_id": "notification_consumer_group_33"}
folder_name = datetime.now().strftime('%Y-%m-%d')
def format_result(consume_message):
    data = json.loads(consume_message[1])
    file_name = datetime.now().strftime("%Y_%m_%d-%I_%M_%S")
    df = pd.DataFrame(data).T #, orient='index'
    df.to_parquet(f'gs://{conf.gcs}/{folder_name}/{file_name}.parquet',
               storage_options={"token": "gcp.json"}, engine='fastparquet')
    print(consume_message)
with beam.Pipeline(options=PipelineOptions()) as p:
    consumer_message = (p | "Reading messages from Kafka" >> kafkaio.KafkaConsume(consumer_config=consumer_config)
                          | 'Windowing' >> beam.WindowInto(window.FixedWindows(30),
                                            trigger=AfterProcessingTime(30),
                                            allowed_lateness=900,
                                           accumulation_mode=AccumulationMode.ACCUMULATING)
                          | 'CombineGlobally' >> beam.Map(format_result))

# window.FixedWindows(30),trigger=beam.transforms.trigger.AfterProcessingTime(30),
# accumulation_mode=beam.transforms.trigger.AccumulationMode.DISCARDING
# allowed_lateness=100,CombineGlobally(format_result).without_defaults() allowed_lateness=30,

Using the code above, a new parquet file is generated for each message. What I would like to do is to group messages by 30 seconds windows and generate one parquet file for each window. I tried different configurations below with no success: beam.CombineGlobally(format_result).without_defaults()) instead of beam.Map(format_result)) beam.ParDo(format_result)) In addition, I have few more questions:

Even though I set the offset by "auto.offset.reset": "earliest", kafka producer starts to read from the last message even if I change the consumer group and can’t figure out why.
Also, I am puzzled by the usage of trigger, allowed_lateness, accumulation_mode. I am not sure if I need them for the this task.

As you can see in the code block above, I also tried using these parameters but it didn’t help.

I searched everywhere but couldn’t find a single example that explains this use case. `

user_949327 · Answer

Here are some changes you should make to your pipeline to get this result:

Remove your trigger if you want a single output per window. Triggers are only needed for getting multiple results per window.
Add a GroupByKey or Combine operation to aggregate the elements. Without such an operation, the windowing has no effect.
I recommend using parquetio from the Beam project itself to ensure you get scalable exactly-once behavior. (See the pydoc from 2.33.0 release)

user_14450057 · Answer

I took a look at the GroupByKey example in the python documentation

Messages I read from KafkaConsumer (I used kafkaio from beam_nuggets.io) have a type of tuple, and in order to use GroupByKey, I tried to create a list in the convert_to_list function by appending the tuples I got from Kafka Consumer. However, GroupByKey still produces no output.

import apache_beam as beam
from beam_nuggets.io import kafkaio

new_list = []
def convert_to_list(consume_message):
    new_list.append(consume_message)
    return new_list


with beam.Pipeline() as pipeline:
    dofn_params = (
            pipeline
            | "Reading messages from Kafka" >> kafkaio.KafkaConsume(consumer_config=consumer_config)
            | 'Fixed 30sec windows' >> beam.WindowInto(beam.window.FixedWindows(30))
            | 'consume message added list' >> beam.ParDo(convert_to_list)
            | 'GroupBykey' >> beam.GroupByKey()
            | 'print' >> beam.Map(print))

I also tried a similar pipeline but this time, I created a list of tuples with beam.Create() instead of reading from kafka, and it works successfully. You can view this pipeline below:

import apache_beam as beam
from beam_nuggets.io import kafkaio


with beam.Pipeline() as pipeline:
    dofn_params = (
            pipeline
            | 'Created Pipeline' >> beam.Create([(None, '{"userId": "921","xx":"123"]),(None, '{"userId": "92111","yy":"123"]))
            | 'Fixed 30sec windows' >> beam.WindowInto(beam.window.FixedWindows(30))
            | 'GroupBykey' >> beam.GroupByKey()
            | 'print' >> beam.Map(print))

I assume the issue in the first approach is related to generating an external list instead of pcollection, but I am not sure. Can you guide me on how to proceed?

Another thing I tried is to use ReadFromKafka function from apache_beam.io.kafka module. But this time I got the following error:

ERROR:apache_beam.utils.subprocess_server:Starting job service with ['java', '-jar', 'user_directory’/.apache_beam/cache/jars\beam-sdks-java-io-expansion-service-2.33.0.jar', '59627']

Java version 11.0.12 is installed on my computer and the ‘java’ command is available.

Writing unique parquet file per windows with Apache Beam Python

2 Answers