You seem to be reading from multiple topics and your timestamp policy is too simplistic (withTimestampFn() was never meant to be public API, I am sending a PR to deprecate it first and then will make it package private). So if you have two topics of different sizes, smaller topic might be consumed really fast pushing your watermark way ahead. This might or might be happening, but this is one of the dangers of using record timestamp for watermark (we should never do that).
On Tue, Nov 6, 2018 at 3:44 AM Kaymak, Tobias <[email protected]> wrote: > Hi, > > I am sharing my experience with you after trying to use the following > pipeline > logic (with Beam 2.6.0 - running on Flink 1.5): > > 1. Reading from KafkaIO, attaching a timestamp from each parsed element > 2. Filtering bad records > 3. Writing to a partitioned table in BigQuery with FILE_LOADS (batch jobs) > every 15 minutes > > I had a working pipeline that does not write to BigQuery directly, but to > Cloud Storage, so it's 3rd step was > > 3. Writing files to GCS in daily "subdirectories" > > I tried to rewrite the pipeline to reduce complexity: Resetting it's state > should no longer be tied to thinking about what to delete on GCS, also > configurable refresh times directly from within the Pipeline was something > I > was looking for. The thing that I needed to change was the output in the > end, > so knew my parsing logic would not change and that should reduce the risk. > > I tested the pipeline within our testcluster and it looked promising. When > I > deployed it last week everything seemed to go smoothly. On Friday I noticed > that I had holes in the data: in the BigQuery tables there were missing > days > (tricky was that the recent days looked fine). (To be sure I reset the > pipeline > and read from the very beginning of each topic from Kafka. Within different > runs, different days were missing.) I spent the weekend rolling back the > changes and trying to figure out what was going on. > > I didn't see any error in the logs (the log level was on WARNING for most > parts), but I thought, well maybe it's because there are too many > partitions > and BigQuery has a limit of 1000 partition operations per day. So I started > reading from just 90 days in the past, but I still had holes (whole days). > > I had a windowing step that I needed for the GCS pipeline, I became aware > that I > wouldn't need this anymore with BigQueryIO so I commented it out and > tested > again, without luck. > > What struck me was that the Flink Cluster didn't do any checkpoints for the > pipeline that was using BigQueryIO - it does so when writing to GCS and I > tested it's failure logic there. Additionally the graph in Flink with > BigQueryIO becomes very complex, but this is something I expected. > > Here is the Pipeline code with the commented out windowing part: > > pipeline > .apply( > KafkaIO.<String, String>read() > .withBootstrapServers(bootstrap) > .withTopics(topics) > .withKeyDeserializer(StringDeserializer.class) > .withValueDeserializer(ConfigurableDeserializer.class) > .updateConsumerProperties( > > ImmutableMap.of(InputMessagesConfig.CONFIG_PROPERTY_NAME, > inputMessagesConfig)) > > .updateConsumerProperties(ImmutableMap.of("auto.offset.reset", "earliest")) > .updateConsumerProperties(ImmutableMap.of("group.id", > "di-beam-consumers")) > > .updateConsumerProperties(ImmutableMap.of("enable.auto.commit", "true")) > .withTimestampPolicyFactory( > TimestampPolicyFactory.withTimestampFn( > new > MessageTimestampExtractor(inputMessagesConfig))) > .withReadCommitted() > .commitOffsetsInFinalize()) > .apply(ParDo.of(new ToEventFn())) > // .apply( > // Window.into(new ZurichTimePartitioningWindowFn()) > // .triggering( > // Repeatedly.forever( > // AfterFirst.of( > // > AfterPane.elementCountAtLeast(bundleSize), > // > AfterProcessingTime.pastFirstElementInPane() > // .plusDelayOf(refreshFrequency)))) > // .withAllowedLateness(Duration.standardDays(1)) > // .discardingFiredPanes()) > .apply( > BigQueryIO.<Event>write() > .withMethod(BigQueryIO.Write.Method.FILE_LOADS) > .withTriggeringFrequency(refreshFrequency) > .withNumFileShards(1) > .to(partitionedTableDynamicDestinations) > .withFormatFunction( > (SerializableFunction<Event, TableRow>) > KafkaToBigQuery::convertUserEventToTableRow) > > .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED) > > .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_TRUNCATE)); > > > I have the feeling that I must make some serious and dumb mistakes as I > know > the Beam framework is very robust. Thanks for taking the time to read this. > > Tobi >
