Sure not a problem. I will open one. Thanks Mohil
On Fri, Jun 26, 2020 at 10:55 AM Luke Cwik <[email protected]> wrote: > Sorry, I didn't open a support case. You should open the support case. > > On Fri, Jun 26, 2020 at 10:41 AM Mohil Khare <[email protected]> wrote: > >> Thanks a lot Luke for following up on this and opening a dataflow >> support. It would be good to know how streamingEngine solved the problem. >> I will really appreciate it if you can share a link for a support case >> once you open it (if it is possible). >> >> Thanks and Regards >> Mohil >> >> >> >> On Fri, Jun 26, 2020 at 8:30 AM Luke Cwik <[email protected]> wrote: >> >>> 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 >>>>>>>>>> >>>>>>>>>> >>>>>>>>>>
