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 <mo...@prosimo.io> 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 <lc...@google.com> 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 <mo...@prosimo.io> 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 <mo...@prosimo.io> 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 <mo...@prosimo.io> 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 <mo...@prosimo.io> 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 <lc...@google.com> 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 <mo...@prosimo.io> >>>>>>> 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 <mo...@prosimo.io> >>>>>>>> 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 >>>>>>>>> >>>>>>>>> >>>>>>>>>
