Questions tagged [apache-beam]

Apache Beam is a unified SDK for batch and stream processing. It allows to specify large-scale data processing workflows with a Beam-specific DSL. Beam workflows can be executed on different runtimes like Apache Flink, Apache Spark, or Google Cloud Dataflow (a cloud service).

Filter by
Sorted by
Tagged with
0 votes
0 answers
21 views

Reading a multi-GB file takes a long time

I'm trying to read an uncompressed 6GB file in a simple pipeline using Apache Beam 2.48.0: with beam.Pipeline(options=pipeline_options) as p: file = p | "read file" >> beam.io....
Grégoire Borel's user avatar
-1 votes
0 answers
15 views

Does Amazon SqsIO.read (inside a Apache Beam) deletes the message from queue after it processing message?

We are developing an application in Apache Beam with runtime as Apache Flink deployed on AWS cloud. In our application we try to read a SQS message (FIFO queue), and spin a new pipeline task. ...
Arun's user avatar
  • 1
0 votes
0 answers
24 views

How can I read from MQTT with the Apache Beam Python SDK

I'm trying to read from an MQTT topic using the Apache Beam Python SDK v2.50.0 with multi-language support - MQTT IO is available via the Java SDK, but not the Python SDK). Here are the docs I'm ...
Daniel Stevenson's user avatar
0 votes
0 answers
11 views

slowly updating global window side input with rate configurable by pipeline option

I'm trying to use a slowly updating global window side inputs to periodically refresh some configuration data in my pipeline p.apply(GenerateSequence.from(0).withRate(1, Duration.standardSeconds(5L))) ...
marengaz's user avatar
  • 1,649
0 votes
1 answer
21 views

Cannot call getSchema when there is no schema when join two Pcollections

I hope all is good for you. I have two PCollection and I want to apply a leftOuterJoin but when I do this apply, I have this error and I don't understand why : java.lang.IllegalStateException: Cannot ...
DiR95's user avatar
  • 49
0 votes
0 answers
15 views

Can't compile a custom Dataflow Classic Template

I'm trying to write a Dataflow Classic Template in Java which I'm not used to. I copied the code from the Bulk Decompress Cloud Storage Files template that decompress gzip files, then I added some ...
kbenzakrilelp's user avatar
0 votes
1 answer
23 views

Unable to use Java Double in @FieldAccess DoFn for fetching nullable DOUBLE schema field

I'm attempting to do the following in my DoFn to extract a double value from an existing nullable field in my schema: @FieldAccess("field1") Double field1 However I'm getting the following ...
Vergo's user avatar
  • 1
0 votes
1 answer
20 views

Using Apache Beam in Python to create a template in GCP Dataflow

Task : I need to take data from Oracle on premise and dump it in BigQuery. Option which I tried There are various ways to achieve the same. I tried first option i.e with Datastream but due to ...
Ashwin Joshi's user avatar
0 votes
0 answers
14 views

Using script_fields with Apache Beam's ElasticsearchIO.read().withQuery()

I'm trying to read data from Elasticsearch using Apache Beam's ElasticsearchIO.read().withQuery(). I want to use a query that includes the script_fields parameter to compute a custom field as part of ...
Rafael Adao's user avatar
0 votes
1 answer
14 views

Question re: FailedInserts PCollection / Beam 2.45 versus 2.49 behavior confirmation

Context: GitHub Issue 26853 I have a question around the above referenced feature request and differences in Beam 2.45 versus 2.49. In case the FailedInserts PCollection is not consumed, BigQueryIO ...
Jdub's user avatar
  • 3
0 votes
1 answer
22 views

Beam Unit Test `assert_that` and `equal_to` doesn't actually compare the result

I noticed that the Beam assert_that and equal_to doesn't seems to be able to compare what I expected. To further testing that, I am testing with a very simple code: import unittest import apache_beam ...
Lydian's user avatar
  • 78
1 vote
1 answer
31 views

GCP PubSub to GCS dataflow

I have one use case to create files in GCS buckets from PubSub with the below conditions: Window size limit Published message size to PubSub topic exceeds 500MB For eg, i) If we specify window size ...
Roshan's user avatar
  • 11
0 votes
0 answers
17 views

Use Avro schema generated during the process in writing operation to Avro file in Apache Beam pipeline/Java

I use Apache Beam 2.49. I have to get data from Oracle database and save it in Avro file on Google Cloud Storage. I know how to get this data, schema and generate GenericRecord from Oracle but I have ...
pijars's user avatar
  • 1
0 votes
0 answers
19 views

Apache beam write PCollection<string,pojo> to GCS bucket

I am trying to put a PCollection<KV<String,POJO>> , I need to write this Pojo to GCS bucket in the avro format, where the string(key) for the map is the GCS folder path that I need to ...
Surya Srivastava's user avatar
0 votes
0 answers
21 views

How to write data to Kinesis Data Stream with Apache Beam using Python

I am trying to build Beam pipeline in Python that reads from and writes to AWS Kinesis Data Streams. I am using the FlinkRunner, with Flink 1.15.4 running locally on my machine. Python 3.8.17 Apache-...
Charl's user avatar
  • 111
1 vote
0 answers
40 views

Apache Beam portable runner to run golang based job on flink

I am trying to learn Apache Beam and trying to create a sample project to learn stream processing. For now, I want to read from a Kafka topic "word" and print the data on console. I have ...
Swapnil1456's user avatar
0 votes
0 answers
40 views

Nested Pipeline - Apache Beam - Flink Runner - Running Local - Failing on Local Cluster

For Context, my use case is to read from an unbounded source, in this case I am using aws SQS. From each message, I am wanting to generate pipelines dynamically that read from another location, ...
Sal Mem's user avatar
0 votes
1 answer
29 views

How do you load a huggingface dataset into an Apache Beam pipeline?

Assuming I have a dataset which is quite large like 'oscar2301'. I have an Apache Beam pipeline that I want to pass the HuggingFace dataset. When I try to pass the dataset to a beam.Create, the code ...
Mohamed Taher Alrefaie's user avatar
0 votes
0 answers
22 views

Join bounded data with unbounded data using CoGroupByKey

I'm joining data from two data sources: one is bounded, using beam.io.ReadFromText. It is a json.gz file one is unbounded, using beam.io.ReadFromPubSub + a Window beam.WindowInto(beam.window.Sessions(...
Grégoire Borel's user avatar
0 votes
0 answers
35 views

Apache Beam only timely ParDo function with no state

I'm trying to write an element to Firestore 1 second after my fixed window closes. I have tried to follow the timely and stateful DoFn guides but they do not have an example of a strictly timely only ...
Joe Moore's user avatar
  • 2,019
1 vote
0 answers
20 views

Google Dataflow flex-template run results in INVALID_ARGUMENT error

I am trying to deploy a Java dataflow job to Google dataflow based on a custom pubsub-to-elasticsearch template. Why is it that this CLI call is successful... --update \ --region us-central1 \ ...
msenkovi's user avatar
0 votes
1 answer
64 views

Beam job run successfully on local machine but failed on dataflow runner

I build a beam job which: Reads data (messages like {"user_id":"u1", "event_name":"logout", "region":"US"}) from pubsub Retrieve data from ...
digitalearth's user avatar
1 vote
0 answers
41 views

How do you trigger a DoFn synchronously in Apache Beam?

The Dataflow pipeline I've written ingests data that has been published synchronously. The order of this data does not matter, but it is published every second. I need to output the result of the ...
Joe Moore's user avatar
  • 2,019
0 votes
0 answers
123 views

Apache Beam not writing data to BigQuery using 2 millions lines file as side input

I'm joining data from two data sources with Apache Beam Python SDK 2.48.0: one is bounded, using beam.io.ReadFromText. It is a json.gz file one is unbounded, using beam.io.ReadFromPubSub + a Window ...
Grégoire Borel's user avatar
1 vote
0 answers
122 views

Apache Beam: Wait N seconds after window closes to execute DoFn

I have real time data being published and read into a Dataflow pipeline in a synchronous manner. I collect the data, window it (1 second fixed) with accumulation then write it to a Firestore DB. This ...
Joe Moore's user avatar
  • 2,019
0 votes
1 answer
39 views

Python, apache beam, kafka and bigquery

I use python and apache beam to read streaming data from kafka and insert the data to big query table but I want to insert the data in batches instead of streaming way. I tried to set pipeline ...
muertto's user avatar
0 votes
1 answer
71 views

Join multiple pCollection in global window

I want to build a Beam program to Streamingly read Pub/Sub messages which is like {"user_id":"u1"} Use the userid to retrieve data from 7+ BigQuery tables. Due to performance ...
digitalearth's user avatar
0 votes
1 answer
34 views

Google dataflow write data to CloudSQL postgres | java.sql.SQLException: Cannot create PoolableConnectionFactory (The connection attempt failed.)

I am trying to create a ETL pipeline using google dataflow to write data to CloudSQL postgres database. But I am facing error connecting to the cloud sql instance. String jdbcConURL = "jdbc:...
Rakesh kumar's user avatar
0 votes
0 answers
32 views

Deploy Scio jobs with custom containers

I would like to deploy a Scio Job on Dataflow with custom containers https://beam.apache.org/documentation/runtime/environments/ Is it possible to package the app with sbt-pack for example, and extend ...
Aurélien V's user avatar
0 votes
1 answer
55 views

OOM Error in Apache beam java streaming pipeline

My dataflow job streaming pipeline is designed as below. Reading updated/inserted row from spanner change stream as datachangerecord if datachangerecord mod is insert than create tablerow if ...
Sweety's user avatar
  • 307
-1 votes
1 answer
109 views

Apache beam python looking up value in other collection

I am trying to lookup a value in a seperate big query table, i need to lookup the link_url in table one using the url in table 2. From table 1 i need just the link_id table 1 link_id link_url id1 ...
Josh Fradley's user avatar
0 votes
0 answers
40 views

Disable Dataflow/Beam logging on FileIO.matchAll()

I have a Dataflow job written in Java (Beam version 2.43.0, Java 11), which reads messages from Pubsub that contains path to Cloud Storage files. Due to high volume of messages, especially during ...
oikonomiyaki's user avatar
  • 7,731
0 votes
0 answers
36 views

Handling the restart with apache beam debezium application

In my current scenario, I have an Apache Beam application that employs Debezium for capturing changes in data. Unlike the typical integration with Kafka, I am not using it in conjunction with Kafka. ...
jps sasadara's user avatar
1 vote
0 answers
24 views

ReadfromKafka stuck and messages are not passed to WriteToBigQuery

I am trying to read a large Kafka topic and insert those message to a BigQuery table from the beginning of the topic using a beam pipeline using Python SDK deployed in GCP Dataflow. Following is my ...
Prasanth Narayanan's user avatar
0 votes
0 answers
32 views

Messed up data in Dataflow python SDK pipeline

In google dataflow with python sdk i have the following pipeline class LogElement(beam.DoFn): def process(self, element): logging.info(f'{element}') yield element class ...
user912830823's user avatar
0 votes
1 answer
26 views

BigQuery table schema types not being translated correctly in apache beam

There is a bug in the python Apache Beam SDK for BigQuery currently which translates BQ TIMESTAMP incorrectly to BQ DATETIME. This seems to have been fixed, but I have a feeling it may be in a pre-...
Joe Moore's user avatar
  • 2,019
0 votes
0 answers
52 views

Apache Beam / GCP Dataflow -- Python sdk harness failed: TypeError: can only concatenate str (not "NoneType") to str

I'm trying to send custom metrics to datadog using a DoFn but the python sdk test harness is failing with an error that I don't know how to interpret: Traceback (most recent call last): File "/...
willwrighteng's user avatar
0 votes
0 answers
21 views

Capturing deserialization exceptions in KafkaIO

I have a typical KafkaIO based source for reading Avro formatted key and value from a Kafka topic. PCollection<KafkaRecord<GenericRecord, GenericRecord>> records = pipeline.apply( ...
pravish's user avatar
  • 33
0 votes
0 answers
40 views

Dataflow FixedWindow triggered more than once during allowed period

I have lots of devices that constantly send me data. I want to use dataflow to parse the data and save it to Firestore. I want to use FirestoreIO for saving. First problem I ran into was that I ...
Virx's user avatar
  • 105
0 votes
0 answers
20 views

universal sentence encoder batch pipeline failing

I have a batch job on DataFlow runner to calculate the embedding from the input text. Through the journey of pipeline. I am using tft.impl.context and impl.AnalyzeAndTransformDataset for the same Here ...
Akarsh Jain's user avatar
0 votes
1 answer
55 views

Dataflow - process single input to multiple outputs using a ptransform

I read data from Pubsub and there are different types of data. I would provide runtime argument based on which it will have to create multiple outputs( using branching) but the idea is i would get ...
Anesh P's user avatar
0 votes
1 answer
129 views

Memory leak in Apache Beam Python ReadFromPubsub IO

Overview We have a Dataflow streaming pipeline that reads messages from a pubsub subscription, transforms the dict to a dataclass and writes the data to postgres. I noticed that occasionally, pubsub ...
leech's user avatar
  • 259
0 votes
0 answers
17 views

How to create Dataflow using Bean that processes three rest calls concurrently

I have three endpoints which I need to query concurrently using Apache Beam in Java. The endpoints are stored in a Set. I would like to create three concurrent Dataflow tasks that query these ...
ZZZ's user avatar
  • 653
0 votes
0 answers
83 views

Dataflow Flex template while running throwing error for all default parameters

I have created a flex template for dataflow and it got build using gcloud successfully but while running the template, it is throwing the error for all default parameters that are supposed to be taken ...
user9851003's user avatar
0 votes
1 answer
47 views

Apache Beam Python pipeline - Wait for second PCollection to be generated

I am reading a GCS bucket and generating a json file which I need to import into BQ later. But I get error "OSError: No files found based on the file pattern" with beam.Pipeline(options=...
user1548923's user avatar
0 votes
0 answers
26 views

Apache beam CustomTimestamp windowing not firing aggregate functions

I'm reading from Kafka topic and applying customtimestamp from the datetime that I receive from the message in kafka, the example Pojo of the message: class Order implements Serializable { String ...
vinayhudli's user avatar
0 votes
1 answer
38 views

GCP Dataflow batch processing

I am doing POC for GCP Dataflow batch processing. I want to pass Pandas Dataframe as batch input and perform columnar transformation and return same batch again. I refer example provided in below link ...
PramodK's user avatar
  • 11
0 votes
0 answers
51 views

DataFlow autoscaling workers not being created

I am trying to run the following beam pipeline in DataFlow and hoping for some help understanding how to optimise it to autoscale. with beam.Pipeline(options=pipeline_options) as p: bq_query = f&...
Danny's user avatar
  • 85
0 votes
0 answers
61 views

Apache Beam dataflow combine per key

I have a problem with my pipeline. My goal is a read around 4k parquet files read it as a numpy array and then make some aggregations eg from one file can make 100 keys each key has some numbers of ...
Dawid 's user avatar
  • 11
0 votes
0 answers
43 views

Overwriting partitions in BigQuery table using dataflow

Im looking for a solution to overwrites multiple partition in BigQuery table using dataflow pipeline I have a pipeline that write data to bq table: input_data = [ {"Date": "2023-01-...
Mikołaj Krawiec's user avatar

1
2 3 4 5
94