It seems as though we have seen this failure before for Dataflow and it was caused because the side input tags needed to be unique in a streaming pipeline.
It looked like this used to be a common occurrence in the Python SDK[1, 2] because it generated tags that weren't unique enough. I would open up a case with Dataflow support with all the information you have provided here. 1: https://issues.apache.org/jira/browse/BEAM-4549 2: https://issues.apache.org/jira/browse/BEAM-4534 On Thu, Jun 25, 2020 at 9:30 PM Mohil Khare <[email protected]> wrote: > Hi Luke and all, > > UPDATE: So when I started my job by *enabling the streaming engine and > keeping the machine type default for the streaming engine (n1-standard-2)*, > the pipeline started successfully. > https://cloud.google.com/dataflow/docs/guides/deploying-a-pipeline#streaming-engine > > Still evaluating to make sure that there is no performance degradation by > doing so. > > Thanks and regards > Mohil > > > On Thu, Jun 25, 2020 at 11:44 AM Mohil Khare <[email protected]> wrote: > >> Hi Luke, >> >> Let me give you some more details about the code. >> As I mentioned before, I am using java sdk 2.19.0 on dataflow. >> >> Default machine type which n1-standard-4. >> Didn't set any numWorkerHarnessThreads (I believe beam/dataflow picks it >> up based on number of cores available) >> >> 1: Code listens for some trigger on pubsub topic: >> /** >> >> * Read From PubSub for topic ANALYTICS_UPDATE and create >> PCollection<String> indicating main pipeline to reload * relevant >> DataAnalyticsData from BQ table */ static class >> MonitorPubSubForDailyAnalyticsDataStatus extends PTransform<PBegin, >> PCollection<POJORepresentingJobCompleteInfo>> { private final String >> subscriptionName; private final String jobProject; >> MonitorPubSubForDailyAnalyticsDataStatus(String subscriptionName, String >> jobProject) { this.subscriptionName = subscriptionName; >> this.jobProject = jobProject; } @Override public >> PCollection<POJORepresentingJobCompleteInfo> expand(PBegin input) { >> return input.getPipeline() .apply("Read_PubSub_Messages", >> PubsubIO.readMessagesWithAttributesAndMessageId().fromSubscription(subscriptionName)) >> .apply("Applying_Windowing", Window.<PubsubMessage>into(new >> GlobalWindows()) >> .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1))) >> .discardingFiredPanes()) .apply("Read_Update_Status", >> ParDo.of(new DoFn<PubsubMessage, POJORepresentingJobCompleteInfo>() { >> @ProcessElement public void >> processElement(@Element PubsubMessage input, >> OutputReceiver<POJORepresentingJobCompleteInfo> out) { >> /*** Read and CReate ***/ >> >> out.output(POJORepresentingJobCompleteInfo); >> } })); } } >> >> 2: Get Latest Updated and Reload new Updates from various BQ tables using >> google cloud bigquery library >> (https://cloud.google.com/bigquery/docs/quickstarts/quickstart-client-libraries) >> >> PCollection<POJORepresentingJobCompleteInfo> analyticsDataStatusUpdates >> = p.apply("Get_Analytics_Data_Status_Updates_pubsub", >> >> new MonitorPubSubForDailyAnalyticsDataStatus(subscriptionName, >> jobProject)); >> >> >> 3. Create various PCollectionViews to be used as side input for decorating >> stream of logs coming from Kafka (will be shown later) >> >> PCollectionView<Map<Stats1Key, Stats1>> Stats1View = >> >> analyticsDataStatusUpdates >> .apply("Reload_Stats1_FromBQ", new ReadStats1()) >> .apply("View_Stats1", View.asSingleton()); >> >> >> PCollectionView<Map<Stats2Key, Stats2>> Stats2View = >> >> analyticsDataStatusUpdates >> .apply("Reload_Stats1_FromBQ", new ReadStats2()) >> .apply("View_Stats1", View.asSingleton()); >> >> . >> >> . >> >> . >> >> . and so one >> >> >> 4. An example of code where we read stats from BQ i.e. in ReadStats1(), >> ReadStats2() and so on >> >> class ReadStatsS1 extends >> PTransform<PCollection<POJORepresentingJobCompleteInfo>, >> PCollection<Map<Stats1Key, Stats1>>> { >> >> @Override public PCollection<Map<Stats1Key, Stats1>> >> expand(PCollection<POJORepresentingJobCompleteInfo> input) { return >> input .apply("Read_From_BigQuery", ParDo.of(new BigQueryRread())) >> .apply("Applying_Windowing", Window.<Map<Stats1Key, >> Stats1>>into(new GlobalWindows()) >> .triggering(Repeatedly.forever(AfterPane.elementCountAtLeast(1))) >> .discardingFiredPanes()); } private class BigQueryRread extends >> DoFn<POJORepresentingJobCompleteInfo, Map<Stats1Key, Stats1>> { >> @ProcessElement public void process(@Element >> POJORepresentingJobCompleteInfo input, ProcessContext c) { >> Map<Stats1Key, Stats1> resultMap = new HashMap<>(); >> try { BigQuery bigQueryClient = >> BigQueryOptions.getDefaultInstance().getService(); String >> sqlQuery = getSqlQuery(input); ///some method to return desired sql query >> based on info present in input QueryJobConfiguration >> queryJobConfiguration = >> QueryJobConfiguration.newBuilder(sqlQuery).setUseLegacySql(false).build(); >> // Create a job ID so that we can safely retry. >> JobId jobId = JobId.of(UUID.randomUUID().toString()); Job >> queryJob = >> bigQueryClient.create(JobInfo.newBuilder(queryJobConfiguration).setJobId(jobId).build()); >> // Wait for the query to complete. queryJob = >> queryJob.waitFor(); if (queryJob == null) { >> logger.p1Error("Big Query Job no longer exists"); } else if >> (queryJob.getStatus().getError() != null) { // You can >> also look at queryJob.getStatus().getExecutionErrors() for all >> // errors, not just the latest one. >> logger.p1Error("Big Query job returned error: {}", >> queryJob.getStatus().getError().toString()); } else { >> //successful case logger.info("Parsing results >> executed by BigQuery"); // Get the results. >> TableResult result = queryJob.getQueryResults(); if >> (null == result || !result.iterateAll().iterator().hasNext()) { >> logger.info("No data found for query: {}", sqlQuery); >> } else { // Print all pages of the results. >> for (FieldValueList row : result.iterateAll()) { >> /*** Parse row and create Stats1Key and Stats from >> that row/ resultMap.put(key, stats); >> } } } >> } } catch (Exception ex) { >> logger.p1Error("Error in executing sql query against Big Query", ex); >> } logger.info("Emitting map of size: {}", resultMap.size()); >> c.output(resultMap); } } >> >> As I mentioned before all classes : ReadStats1(), ReadStats2() etc >> follow above code design >> >> >> 5. Using KafkaIO we read continuous stream of data from kafka >> >> PCollection<POJO> Logs = >> >> p .apply("Read__Logs_From_Kafka", KafkaIO.<String, >> byte[]>read() .withBootstrapServers(String.join(",", >> bootstrapServerToConnectTo)) .withTopic("topic") >> .withKeyDeserializer(StringDeserializer.class) >> .withValueDeserializer(ByteArrayDeserializer.class) >> .withConsumerConfigUpdates(kafkaConsumerProperties) >> .withConsumerFactoryFn(consumerFactoryObj) >> .commitOffsetsInFinalize()) .apply("Applying_Fixed_Window_Logs", >> Window.<KafkaRecord<String, >> byte[]>>into(FixedWindows.of(Duration.standardSeconds(10))) >> .triggering(Repeatedly.forever(AfterWatermark.pastEndOfWindow().withEarlyFirings(AfterPane.elementCountAtLeast(1)))) >> .withAllowedLateness(Duration.standardDays(1)) >> .discardingFiredPanes()) >> .apply("Convert_KafkaRecord_To_PCollection<POJO>", >> ParDo.of(new ParseLogs()); >> >> >> 6. Take these logs and apply another Transform providing aforementioned BQ >> reads as side input i.e. something like this >> >> Logs.apply("decorate", new Decorate().withSideInput(Stats1View, >> Stats2View...); >> >> >> Please Note: I tried commenting out code where I added side input to the >> above transform and still landed up in the same crash. So Issue is >> definitely in adding >> >> more than a certain number of PCollectionView transforms. I already had 3-4 >> such transforms and it was working fine. Yesterday I added a few more and >> started seeing crashes. >> >> If I enable just one of the newly added PCollectionView transforms (keeping >> old 3-4 intact), then everything works fine. Moment I enable another new >> transform, a crash happens. >> >> >> Hope this provides some more insight. Let me know if I need to open a ticket >> or I am doing something wrong or some extra configuration or different >> worker machine type need to be provided. >> >> >> Thanks and Regards >> >> Mohil >> >> >> On Thu, Jun 25, 2020 at 11:01 AM Mohil Khare <[email protected]> wrote: >> >>> Hi Luke, >>> >>> I can send you a code snippet with more details if it helps. >>> >>> BTW found similar issue here: >>> http://mail-archives.apache.org/mod_mbox/beam-user/201801.mbox/%3ccaf9t7_74pkr7fj51-6_tbsycz9aiz_xsm7rcali5kmkd1ng...@mail.gmail.com%3E >>> >>> Thanks and Regards >>> Mohil >>> >>> On Thu, Jun 25, 2020 at 10:58 AM Mohil Khare <[email protected]> wrote: >>> >>>> Hi Luke, >>>> Thanks for your response, I tried looking at worker logs using the >>>> logging service of GCP and unable to get a clear picture. Not sure if its >>>> due to memory pressure or low number of harness threads. >>>> Attaching a few more screenshots of crash logs that I found as wells >>>> json dump of logs. >>>> >>>> Let me know if you still think opening a ticket is a right way to go. >>>> >>>> Thanks and regards >>>> Mohil >>>> >>>> On Thu, Jun 25, 2020 at 10:00 AM Luke Cwik <[email protected]> wrote: >>>> >>>>> Try looking at the worker logs to get a full stack trace. Take a look >>>>> at this page for some debugging guidance[1] or consider opening a support >>>>> case with GCP. >>>>> >>>>> 1: >>>>> https://cloud.google.com/dataflow/docs/guides/troubleshooting-your-pipeline >>>>> >>>>> On Thu, Jun 25, 2020 at 1:42 AM Mohil Khare <[email protected]> wrote: >>>>> >>>>>> BTW, just to make sure that there is no issue with any individual >>>>>> PTransform, I enabled each one of them one by one and the pipeline >>>>>> started >>>>>> successfully. Issue happens as soon as I enable more than one new >>>>>> aforementioned PTransform. >>>>>> >>>>>> Thanks and regards >>>>>> Mohil >>>>>> >>>>>> On Thu, Jun 25, 2020 at 1:26 AM Mohil Khare <[email protected]> wrote: >>>>>> >>>>>>> Hello All, >>>>>>> >>>>>>> I am using the BEAM java 2.19.0 version on google dataflow. >>>>>>> >>>>>>> Need urgent help in debugging one issue. >>>>>>> >>>>>>> I recently added 3-4 new PTransformations. to an existing pipeline >>>>>>> where I read data from BQ for a certain timestamp and create >>>>>>> PCollectionView<Map<Key,value>> to be used as side input in other >>>>>>> PTransforms. >>>>>>> >>>>>>> i.e. something like this: >>>>>>> >>>>>>> /** >>>>>>> * Get PCollectionView Stats1 >>>>>>> */ >>>>>>> PCollectionView<Map<Stats1Key, Stats1>> stats1View = >>>>>>> jobCompleteStatus >>>>>>> .apply("Reload_MonthlyS2Stats_FromBQ", new ReadStatsS1()) >>>>>>> .apply("View_S1STATS", View.asSingleton()); >>>>>>> >>>>>>> /** >>>>>>> * Get PCollectionView of Stats2 >>>>>>> */ >>>>>>> PCollectionView<Map<Stats2Key, Stats2>> stats2View = >>>>>>> jobCompleteStatus >>>>>>> .apply("Reload_OptimalAppCharsInfo_FromBQ", new ReadStatsS2()) >>>>>>> .apply("View_S2STATS", View.asSingleton()); >>>>>>> >>>>>>> >>>>>>> and a couple more like these PTransforms. Here jobCompleteStatus is a >>>>>>> message >>>>>>> >>>>>>> received from PubSub that act as a trigger to reload these views. >>>>>>> >>>>>>> The moment I deployed the above pipeline, it didn't start and >>>>>>> >>>>>>> error reporting gave weird exceptions(see attached screenshot1 and >>>>>>> screenshot) which I don't know how to debug. >>>>>>> >>>>>>> >>>>>>> Then as an experiment I made a change where I enabled only one new >>>>>>> transformation >>>>>>> >>>>>>> and disabled others. This time I didn't see any issue. >>>>>>> >>>>>>> So it looks like some memory issue. >>>>>>> >>>>>>> I also compared worker logs between working case and non working case >>>>>>> >>>>>>> and it looks resources were not granted in non working case. >>>>>>> >>>>>>> (See attached working-workerlogs and nonworking-workerlogs) >>>>>>> >>>>>>> I could't find any other log. >>>>>>> >>>>>>> >>>>>>> I would really appreciate quick help here. >>>>>>> >>>>>>> >>>>>>> Thanks and Regards >>>>>>> >>>>>>> Mohil >>>>>>> >>>>>>> >>>>>>>
