Serverless Data Processing with Dataflow - Advanced Streaming Analytics Pipeline with Dataflow (Python) Reviews

NOUREDDINE E. · Reviewed about 4 years ago

Errori Python. segnalata issue in github

Fabrizio P. · Reviewed about 4 years ago

Errori Python irrisolvibili. (df-env) (base) jupyter@theia-20220321-123339:~/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab$ python3 streaming_minute_traffic_pipeline.py --project=${PROJECT_ID} --region=${REGION} --staging_location=${PIPELINE_FOLDER}/staging --temp_location=${PIPELINE_FOLDER}/temp --runner=${RUNNER} --input_topic=${PUBSUB_TOPIC} --window_duration=${WINDOW_DURATION} --allowed_lateness=${ALLOWED_LATENESS} --table_name=${OUTPUT_TABLE_NAME} --dead_letter_bucket=${DEADLETTER_BUCKET} streaming_minute_traffic_pipeline.py:116: FutureWarning: WriteToFiles is experimental. max_writers_per_bundle=0) /home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/io/fileio.py:550: BeamDeprecationWarning: options is deprecated since First stable release. References to <pipeline>.options will not be supported p.options.view_as(GoogleCloudOptions).temp_location or Traceback (most recent call last): File "streaming_minute_traffic_pipeline.py", line 140, in <module> run() File "streaming_minute_traffic_pipeline.py", line 116, in run max_writers_per_bundle=0) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pvalue.py", line 137, in __or__ return self.pipeline.apply(ptransform, self) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pipeline.py", line 652, in apply transform.transform, pvalueish, label or transform.label) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pipeline.py", line 662, in apply return self.apply(transform, pvalueish) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pipeline.py", line 708, in apply pvalueish_result = self.runner.apply(transform, pvalueish, self._options) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/runners/dataflow/dataflow_runner.py", line 141, in apply return super().apply(transform, input, options) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/runners/runner.py", line 185, in apply return m(transform, input, options) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/runners/runner.py", line 215, in apply_PTransform return transform.expand(input) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/io/fileio.py", line 577, in expand | beam.ParDo( File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pvalue.py", line 137, in __or__ return self.pipeline.apply(ptransform, self) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pipeline.py", line 652, in apply transform.transform, pvalueish, label or transform.label) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pipeline.py", line 662, in apply return self.apply(transform, pvalueish) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/pipeline.py", line 708, in apply pvalueish_result = self.runner.apply(transform, pvalueish, self._options) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/runners/dataflow/dataflow_runner.py", line 141, in apply return super().apply(transform, input, options) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/runners/runner.py", line 185, in apply return m(transform, input, options) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/runners/dataflow/dataflow_runner.py", line 830, in apply_GroupByKey return transform.expand(pcoll) File "/home/jupyter/project/training-data-analyst/quests/dataflow_python/7_Advanced_Streaming_Analytics/lab/df-env/lib/python3.7/site-packages/apache_beam/transforms/core.py", line 2609, in expand raise ValueError(msg) ValueError: GroupRecordsByDestinationAndShard: Unsafe trigger: `AfterProcessingTime(delay=120)` may lose data. Reason: MAY_FINISH. This can be overriden with the --allow_unsafe_triggers flag. (df-env) (base)

Fabrizio P. · Reviewed about 4 years ago

Ray K. · Reviewed about 4 years ago

Edyta K. · Reviewed about 4 years ago

Rubén L. · Reviewed about 4 years ago

Ehsan U. · Reviewed about 4 years ago

Rubén L. · Reviewed about 4 years ago

Jacob V. · Reviewed about 4 years ago

Richard K. · Reviewed about 4 years ago

Irene K. · Reviewed about 4 years ago

Excellent level of detail and description.

Tristan T. · Reviewed about 4 years ago

shaik T. · Reviewed about 4 years ago

Marek D. · Reviewed about 4 years ago

Python run command for the pipeline gives an error. Requires an extra flag to run

Berkay B. · Reviewed about 4 years ago

Brady B. · Reviewed about 4 years ago

Alexandre S. · Reviewed about 4 years ago

Abhishek C. · Reviewed about 4 years ago

few stuff to modify to have the results but ok

Charles M. · Reviewed about 4 years ago

Zelmar M. · Reviewed about 4 years ago

There are a couple of problems with this lab. First, when running the pipeline I got the following error: ValueError: GroupRecordsByDestinationAndShard: Unsafe trigger: `AfterProcessingTime(delay=120)` may lose data. Reason: MAY_FINISH. This can be overriden with the --allow_unsafe_triggers flag. I was able to fix this by changing this: trigger=AfterProcessingTime(120) to this: trigger=Repeatedly(AfterProcessingTime(120)) Second, we are asked to add the following to ConvertToCommonLogFn: yield beam.pvalue.TaggedOutput('unparsed_row', element.decode('utf-8')) However the lab code already includes the following in the previous line: element = element.decode('utf-8') If this isn't removed, we get the following error due to double-decoding: AttributeError: 'str' object has no attribute 'decode' [while running 'ParseJson/ParDo(ConvertToCommonLogFn)-ptransform-81']

Robert L. · Reviewed about 4 years ago

Hanh T. · Reviewed about 4 years ago

Edward U. · Reviewed about 4 years ago

Jack R. · Reviewed about 4 years ago

Thiago M. · Reviewed about 4 years ago