You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "charlespnh (via GitHub)" <gi...@apache.org> on 2025/06/19 14:51:42 UTC
[PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
charlespnh opened a new pull request, #35375:
URL: https://github.com/apache/beam/pull/35375
**Please** add a meaningful description for your change here
Part of a larger effort https://github.com/apache/beam/issues/35069 and https://github.com/apache/beam/issues/35068 to add more examples involving Kafka and ML use cases.
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3023760342
Reminder, please take a look at this pr: @jrmccluskey
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3040115483
Retest this please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207668253
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
Review Comment:
After previous removals of non-printable ascii characters, it maybe that some comments then become empty. That row is invalid data, since there isn't comment to do inference on... Or maybe I'm missing sth here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207706730
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
+ # This pipeline makes the same point by converting all words
+ # into lowercase.
+ - type: MapToFields
+ name: LowerCase
+ input: Truncating
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text: "comment_text.lower()"
+ likes: likes
+ replies: replies
+
+ # With VertexAIModelHandlerJSON model handler,
+ # RunInference transform performs remote inferences by
+ # sending POST requests to the Vertex AI endpoint that
+ # our distilbert-base-uncased model is being deployed to.
+ - type: RunInference
+ name: DistilBERTRemoteInference
+ input: LowerCase
+ config:
+ inference_tag: "inference"
+ model_handler:
+ type: "VertexAIModelHandlerJSON"
+ config:
+ endpoint_id: "{{ ENDPOINT }}"
+ project: "{{ PROJECT }}"
+ location: "{{ LOCATION }}"
+ preprocess:
+ callable: 'lambda x: x.comment_text'
+
+ # Parse inference results output
+ - type: MapToFields
+ name: FormatInferenceOutput
+ input: DistilBERTRemoteInference
+ config:
+ language: python
+ fields:
+ video_id:
+ expression: video_id
+ output_type: string
+ comment_text:
+ callable: "lambda x: x.comment_text"
+ output_type: string
+ label:
+ callable: "lambda x: x.inference.inference[0]['label']"
+ output_type: string
+ score:
+ callable: "lambda x: x.inference.inference[0]['score']"
+ output_type: number
+ likes:
+ expression: likes
+ output_type: integer
+ replies:
+ expression: replies
+ output_type: integer
+
+ # Assign windows to each element of the unbounded PCollection.
+ - type: WindowInto
+ name: Windowing
+ input: FormatInferenceOutput
+ config:
+ windowing:
+ type: fixed
+ size: 30s
+
+ # Write all inference results to a BigQuery table.
+ - type: WriteToBigQuery
+ name: WriteInferenceResultsToBQ
+ input: Windowing
+ config:
+ table: "{{ PROJECT }}.{{ DATASET }}.{{ TABLE }}"
+ create_disposition: CREATE_IF_NEEDED
+ write_disposition: WRITE_APPEND
+
+options:
+ yaml_experimental_features: ML
+
+# Expected:
+# Row(video_id='XpVt6Z1Gjjo', comment_text='I AM HAPPY', likes=1, replies=1)
+# Row(video_id='XpVt6Z1Gjjo', comment_text='I AM SAD', likes=1, replies=1)
Review Comment:
This is actually part of the unit test with the way the test suit is set up https://github.com/apache/beam/blob/master/sdks/python/apache_beam/yaml/examples/testing/examples_test.py#L202-L216
This test suit checks for all yaml files in the directory and run the unit test, so if a yaml pipeline is added without properly adding the unit test to this test suit it will fail. This test suit is also run as part of CI/CD, hence why unit test and mocking are implemented over here instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-2988439288
In order for the example to be self-contained as much as possible, I decided to have the pipeline to both pushes the data from GCS to a Kafka topic and also reads from that same topic. It won't be a proper streaming pipeline, but the user doesn't have to separately do additional work to push the data to the Kafka topic before being able to finally run the example.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207676118
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
Review Comment:
AFAIK that's the approach. (See for example https://github.com/apache/beam/blob/master/sdks/python/apache_beam/examples/inference/pytorch_sentiment.py#L71-L72)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207707678
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -782,9 +853,67 @@ def _db_io_read_test_processor(
return test_spec
+@YamlExamplesTestSuite.register_test_preprocessor(
Review Comment:
See previous comment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2222705438
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -140,15 +157,62 @@ def test_pubsub_read(
pubsub_messages = input_data.pubsub_messages_data()
return (
- pbegin
+ pcoll
| beam.Create([json.loads(msg.data) for msg in pubsub_messages])
| beam.Map(lambda element: beam.Row(**element)))
+@beam.ptransform.ptransform_fn
+def test_run_inference(pcoll, inference_tag, model_handler):
+ """
+ This PTransform simulates the behavior of the RunInference transform.
+
+ Args:
+ pcoll: The input PCollection.
+ inference_tag: The tag to use for the returned inference.
+ model_handler: A configuration for the respective ML model handler
+
+ Returns:
+ A PCollection containing the enriched data.
+ """
+
+ from apache_beam.ml.inference.base import PredictionResult
+ from apache_beam.typehints.row_type import RowTypeConstraint
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207669136
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2211150667
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -140,15 +157,62 @@ def test_pubsub_read(
pubsub_messages = input_data.pubsub_messages_data()
return (
- pbegin
+ pcoll
| beam.Create([json.loads(msg.data) for msg in pubsub_messages])
| beam.Map(lambda element: beam.Row(**element)))
+@beam.ptransform.ptransform_fn
+def test_run_inference(pcoll, inference_tag, model_handler):
+ """
+ This PTransform simulates the behavior of the RunInference transform.
+
+ Args:
+ pcoll: The input PCollection.
+ inference_tag: The tag to use for the returned inference.
+ model_handler: A configuration for the respective ML model handler
+
+ Returns:
+ A PCollection containing the enriched data.
+ """
+
+ from apache_beam.ml.inference.base import PredictionResult
+ from apache_beam.typehints.row_type import RowTypeConstraint
Review Comment:
Lets put the imports at the top
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -140,15 +157,62 @@ def test_pubsub_read(
pubsub_messages = input_data.pubsub_messages_data()
return (
- pbegin
+ pcoll
| beam.Create([json.loads(msg.data) for msg in pubsub_messages])
| beam.Map(lambda element: beam.Row(**element)))
+@beam.ptransform.ptransform_fn
+def test_run_inference(pcoll, inference_tag, model_handler):
+ """
+ This PTransform simulates the behavior of the RunInference transform.
+
+ Args:
+ pcoll: The input PCollection.
+ inference_tag: The tag to use for the returned inference.
+ model_handler: A configuration for the respective ML model handler
+
+ Returns:
+ A PCollection containing the enriched data.
+ """
+
+ from apache_beam.ml.inference.base import PredictionResult
+ from apache_beam.typehints.row_type import RowTypeConstraint
Review Comment:
Lets put the imports at the top of file
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2199386982
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
Review Comment:
Ditto regarding secret managers
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -114,14 +114,31 @@ def _fn(row):
@beam.ptransform.ptransform_fn
def test_kafka_read(
- pcoll,
+ pbegin,
Review Comment:
This should either be a valid PCollection or a pipeline object (not pbegin).
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -782,9 +853,67 @@ def _db_io_read_test_processor(
return test_spec
+@YamlExamplesTestSuite.register_test_preprocessor(
Review Comment:
Is it possible to add these as tests embedded in the YAML file under a "tests:" section instead of being implemented in Python ?
@derrickaw might know more.
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
Review Comment:
Is there a better approach than just dropping extra tokens here ? @damccorm may have input.
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
+ # This pipeline makes the same point by converting all words
+ # into lowercase.
+ - type: MapToFields
+ name: LowerCase
+ input: Truncating
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text: "comment_text.lower()"
+ likes: likes
+ replies: replies
+
+ # With VertexAIModelHandlerJSON model handler,
+ # RunInference transform performs remote inferences by
+ # sending POST requests to the Vertex AI endpoint that
+ # our distilbert-base-uncased model is being deployed to.
+ - type: RunInference
+ name: DistilBERTRemoteInference
+ input: LowerCase
+ config:
+ inference_tag: "inference"
+ model_handler:
+ type: "VertexAIModelHandlerJSON"
+ config:
+ endpoint_id: "{{ ENDPOINT }}"
+ project: "{{ PROJECT }}"
+ location: "{{ LOCATION }}"
+ preprocess:
+ callable: 'lambda x: x.comment_text'
+
+ # Parse inference results output
+ - type: MapToFields
+ name: FormatInferenceOutput
+ input: DistilBERTRemoteInference
+ config:
+ language: python
+ fields:
+ video_id:
+ expression: video_id
+ output_type: string
+ comment_text:
+ callable: "lambda x: x.comment_text"
+ output_type: string
+ label:
+ callable: "lambda x: x.inference.inference[0]['label']"
Review Comment:
Does the inference list always return one index ?
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
Review Comment:
"does not distinguish between upper and lower case tokens"
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
Review Comment:
Ditto. This might be too restrictive and may drop valid data.
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
Review Comment:
So we are filtering out records that has None or any field ? Can we include such records with default values for missing fields or some representation of None ?
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
Review Comment:
May users of the pipeline expect to see such characters ? Can we preserve using a different character encoding ?
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
Review Comment:
Can we make this part pluggable so that someone who has a true Kafka topic with valid values we read here can just use that to execute a true streaming pipeline ?
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
Review Comment:
This might be too restrictive for someone using such a pipeline in practice.
##########
sdks/python/apache_beam/yaml/examples/README.md:
##########
@@ -231,8 +231,9 @@ gcloud dataflow yaml run $JOB_NAME \
### ML
-These examples leverage the built-in `Enrichment` transform for performing
-ML enrichments.
+These examples include the built-in `Enrichment` transform for performing
Review Comment:
Probably better to list the examples here since we'll add more.
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -114,14 +114,31 @@ def _fn(row):
@beam.ptransform.ptransform_fn
def test_kafka_read(
- pcoll,
+ pbegin,
format,
topic,
bootstrap_servers,
auto_offset_reset_config,
consumer_config):
+ """
+ This PTransform simulates the behavior of the ReadFromKafka transform
+ with the RAW format by simply using some fixed sample text data and
+ encode it to raw bytes.
+
+ Args:
+ pbegin: The input PCollection.
+ format:
Review Comment:
Pls add descriptions for other fields.
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
Review Comment:
We should update this to use secret managers [1] when it's available. @damccorm is this at a state so that we can try (or is there an ETA) ?
[1] https://docs.google.com/document/d/1Ng00Kw-vnG9kKd3RteC6Q5_YN8yDtXRDHkn570GPQrY/edit?tab=t.0#heading=h.c0uts5ftkk58
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
+ # This pipeline makes the same point by converting all words
+ # into lowercase.
+ - type: MapToFields
+ name: LowerCase
+ input: Truncating
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text: "comment_text.lower()"
+ likes: likes
+ replies: replies
+
+ # With VertexAIModelHandlerJSON model handler,
+ # RunInference transform performs remote inferences by
+ # sending POST requests to the Vertex AI endpoint that
+ # our distilbert-base-uncased model is being deployed to.
+ - type: RunInference
+ name: DistilBERTRemoteInference
+ input: LowerCase
+ config:
+ inference_tag: "inference"
+ model_handler:
+ type: "VertexAIModelHandlerJSON"
+ config:
+ endpoint_id: "{{ ENDPOINT }}"
+ project: "{{ PROJECT }}"
+ location: "{{ LOCATION }}"
+ preprocess:
+ callable: 'lambda x: x.comment_text'
+
+ # Parse inference results output
+ - type: MapToFields
+ name: FormatInferenceOutput
+ input: DistilBERTRemoteInference
+ config:
+ language: python
+ fields:
+ video_id:
+ expression: video_id
+ output_type: string
+ comment_text:
+ callable: "lambda x: x.comment_text"
+ output_type: string
+ label:
+ callable: "lambda x: x.inference.inference[0]['label']"
+ output_type: string
+ score:
+ callable: "lambda x: x.inference.inference[0]['score']"
+ output_type: number
+ likes:
+ expression: likes
+ output_type: integer
+ replies:
+ expression: replies
+ output_type: integer
+
+ # Assign windows to each element of the unbounded PCollection.
+ - type: WindowInto
+ name: Windowing
+ input: FormatInferenceOutput
+ config:
+ windowing:
+ type: fixed
+ size: 30s
+
+ # Write all inference results to a BigQuery table.
+ - type: WriteToBigQuery
+ name: WriteInferenceResultsToBQ
+ input: Windowing
+ config:
+ table: "{{ PROJECT }}.{{ DATASET }}.{{ TABLE }}"
+ create_disposition: CREATE_IF_NEEDED
+ write_disposition: WRITE_APPEND
+
+options:
+ yaml_experimental_features: ML
+
+# Expected:
+# Row(video_id='XpVt6Z1Gjjo', comment_text='I AM HAPPY', likes=1, replies=1)
+# Row(video_id='XpVt6Z1Gjjo', comment_text='I AM SAD', likes=1, replies=1)
Review Comment:
Pls delete commented values and probably move to actual expected values under a "tests:" section.
##########
sdks/python/apache_beam/yaml/examples/README.md:
##########
@@ -231,8 +231,9 @@ gcloud dataflow yaml run $JOB_NAME \
### ML
-These examples leverage the built-in `Enrichment` transform for performing
-ML enrichments.
+These examples include the built-in `Enrichment` transform for performing
Review Comment:
Probably better to list the examples here since we'll add more.
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -145,10 +162,57 @@ def test_pubsub_read(
| beam.Map(lambda element: beam.Row(**element)))
+@beam.ptransform.ptransform_fn
Review Comment:
Is it possible to test with a mock model hander instead of completely mocking the transform here ?
cc: @damccorm
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
+ # This pipeline makes the same point by converting all words
+ # into lowercase.
+ - type: MapToFields
+ name: LowerCase
+ input: Truncating
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text: "comment_text.lower()"
+ likes: likes
+ replies: replies
+
+ # With VertexAIModelHandlerJSON model handler,
+ # RunInference transform performs remote inferences by
+ # sending POST requests to the Vertex AI endpoint that
+ # our distilbert-base-uncased model is being deployed to.
+ - type: RunInference
+ name: DistilBERTRemoteInference
+ input: LowerCase
+ config:
+ inference_tag: "inference"
+ model_handler:
+ type: "VertexAIModelHandlerJSON"
+ config:
+ endpoint_id: "{{ ENDPOINT }}"
+ project: "{{ PROJECT }}"
+ location: "{{ LOCATION }}"
+ preprocess:
+ callable: 'lambda x: x.comment_text'
+
+ # Parse inference results output
+ - type: MapToFields
+ name: FormatInferenceOutput
+ input: DistilBERTRemoteInference
+ config:
+ language: python
+ fields:
+ video_id:
+ expression: video_id
+ output_type: string
+ comment_text:
+ callable: "lambda x: x.comment_text"
+ output_type: string
+ label:
+ callable: "lambda x: x.inference.inference[0]['label']"
+ output_type: string
+ score:
+ callable: "lambda x: x.inference.inference[0]['score']"
+ output_type: number
+ likes:
+ expression: likes
+ output_type: integer
+ replies:
+ expression: replies
+ output_type: integer
+
+ # Assign windows to each element of the unbounded PCollection.
+ - type: WindowInto
+ name: Windowing
+ input: FormatInferenceOutput
+ config:
+ windowing:
+ type: fixed
+ size: 30s
+
+ # Write all inference results to a BigQuery table.
+ - type: WriteToBigQuery
+ name: WriteInferenceResultsToBQ
+ input: Windowing
+ config:
+ table: "{{ PROJECT }}.{{ DATASET }}.{{ TABLE }}"
+ create_disposition: CREATE_IF_NEEDED
+ write_disposition: WRITE_APPEND
+
+options:
+ yaml_experimental_features: ML
+
+# Expected:
Review Comment:
Can we add some YAML tests by mocking the Kafka, GCS transforms ? RunInference also can be mocked or changed to use some local model handler.
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/README.md:
##########
@@ -0,0 +1,95 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# Inference Examples Catalog
+
+<!-- TOC -->
+
+* [Inference Examples Catalog](#examples-catalog)
+ * [Streaming Sentiment Analysis](#streaming-sentiment-analysis)
+
+<!-- TOC -->
+
+## Streaming Sentiment Analysis
+
+The example leverages the `RunInference` transform with Vertex AI
+model handler [VertexAIModelHandlerJSON](
+https://beam.apache.org/releases/pydoc/current/apache_beam.yaml.yaml_ml#apache_beam.yaml.yaml_ml.VertexAIModelHandlerJSONProvider),
+in addition to Kafka IO to demonstrate an end-to-end example of a
+streaming sentiment analysis pipeline. The dataset to perform
+sentiment analysis on is the YouTube video comments and can be found
+on Kaggle [here](
+https://www.kaggle.com/datasets/datasnaek/youtube?select=UScomments.csv).
+
+Download the dataset and copy over to a GCS bucket:
+```sh
+gcloud storage cp /path/to/UScomments.csv gs://YOUR_BUCKET/UScomments.csv
+```
+
+For setting up Kafka, an option is to use [Click to Deploy](
Review Comment:
Might be good to introduce a script that sets up resources for executing the pipeline if possible. Beam already has tests that startup a Docker-based Kafka in GCP and create BQ datasets.
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
+ # This pipeline makes the same point by converting all words
+ # into lowercase.
+ - type: MapToFields
+ name: LowerCase
+ input: Truncating
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text: "comment_text.lower()"
+ likes: likes
+ replies: replies
+
+ # With VertexAIModelHandlerJSON model handler,
+ # RunInference transform performs remote inferences by
+ # sending POST requests to the Vertex AI endpoint that
+ # our distilbert-base-uncased model is being deployed to.
+ - type: RunInference
+ name: DistilBERTRemoteInference
+ input: LowerCase
+ config:
+ inference_tag: "inference"
+ model_handler:
+ type: "VertexAIModelHandlerJSON"
+ config:
+ endpoint_id: "{{ ENDPOINT }}"
+ project: "{{ PROJECT }}"
+ location: "{{ LOCATION }}"
+ preprocess:
+ callable: 'lambda x: x.comment_text'
+
+ # Parse inference results output
+ - type: MapToFields
+ name: FormatInferenceOutput
+ input: DistilBERTRemoteInference
+ config:
+ language: python
+ fields:
+ video_id:
+ expression: video_id
+ output_type: string
+ comment_text:
+ callable: "lambda x: x.comment_text"
+ output_type: string
+ label:
+ callable: "lambda x: x.inference.inference[0]['label']"
+ output_type: string
+ score:
+ callable: "lambda x: x.inference.inference[0]['score']"
Review Comment:
Ditto.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3089916966
For future references, Dataflow job for this pipeline run can be found here https://console.cloud.google.com/dataflow/jobs/us-central1/2025-07-18_07_34_58-7493184517937847816;bottomTab=WORKER_LOGS;expandBottomPanel=false;graphView=0;logsSeverity=INFO;step=?pageState=(%22dfTime%22:(%22s%22:%222025-07-18T14:34:59.391Z%22,%22e%22:%222025-07-18T15:48:00.798Z%22))&project=apache-beam-testing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3101210568
Also, ML test suite seems to be green at HEAD ?
https://github.com/apache/beam/actions/workflows/beam_PreCommit_Python_ML.yml
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207633046
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
Review Comment:
You can introduce conditional branch with jinja templatization [1] by adding, for example,
```
{% if true_streaming == "true" %} ... {% endif %}
```
... but testing is no longer straightforward with how the current test suit is set up (you have to start passing jinja variables in addition to this yaml pipeline)...
[1] https://beamsummit.org/slides/2024/BeamYAML_Advancedtopics.pdf
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207722726
##########
sdks/python/apache_beam/yaml/examples/README.md:
##########
@@ -231,8 +231,9 @@ gcloud dataflow yaml run $JOB_NAME \
### ML
-These examples leverage the built-in `Enrichment` transform for performing
-ML enrichments.
+These examples include the built-in `Enrichment` transform for performing
Review Comment:
Added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207686093
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
+ pattern = re.compile(fr"[^{allowed}]")
+ return pattern.sub("", row.comment_text).strip()
+ likes: likes
+ replies: replies
+
+ # Remove rows that have empty comment text
+ # after previously removing unexpected characters.
+ - type: Filter
+ name: FilterForProperComments
+ input: RemoveWeirdCharacters
+ config:
+ language: python
+ keep:
+ callable: |
+ def filter(row):
+ return len(row.comment_text) > 0
+
+ # HuggingFace's distilbert-base-uncased is used for inference,
+ # which accepts string with a maximum limit of 250 tokens.
+ # Some of the comment strings can be large and are well over
+ # this limit after tokenization.
+ # This transform truncates the comment string and ensure
+ # every comment satisfy the maximum token limit.
+ - type: MapToFields
+ name: Truncating
+ input: FilterForProperComments
+ config:
+ language: python
+ dependencies:
+ - 'transformers>=4.48.0,<4.49.0'
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased", use_fast=True)
+
+ def truncate_sentence(row):
+ tokens = tokenizer.tokenize(row.comment_text)
+ if len(tokens) >= 250:
+ tokens = tokens[:250]
+ truncated_sentence = tokenizer.convert_tokens_to_string(tokens)
+ else:
+ truncated_sentence = row.comment_text
+
+ return truncated_sentence
+ likes: likes
+ replies: replies
+
+ # HuggingFace's distilbert-base-uncased does not distinguish
+ # between 'english' and 'English'.
+ # This pipeline makes the same point by converting all words
+ # into lowercase.
+ - type: MapToFields
+ name: LowerCase
+ input: Truncating
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text: "comment_text.lower()"
+ likes: likes
+ replies: replies
+
+ # With VertexAIModelHandlerJSON model handler,
+ # RunInference transform performs remote inferences by
+ # sending POST requests to the Vertex AI endpoint that
+ # our distilbert-base-uncased model is being deployed to.
+ - type: RunInference
+ name: DistilBERTRemoteInference
+ input: LowerCase
+ config:
+ inference_tag: "inference"
+ model_handler:
+ type: "VertexAIModelHandlerJSON"
+ config:
+ endpoint_id: "{{ ENDPOINT }}"
+ project: "{{ PROJECT }}"
+ location: "{{ LOCATION }}"
+ preprocess:
+ callable: 'lambda x: x.comment_text'
+
+ # Parse inference results output
+ - type: MapToFields
+ name: FormatInferenceOutput
+ input: DistilBERTRemoteInference
+ config:
+ language: python
+ fields:
+ video_id:
+ expression: video_id
+ output_type: string
+ comment_text:
+ callable: "lambda x: x.comment_text"
+ output_type: string
+ label:
+ callable: "lambda x: x.inference.inference[0]['label']"
Review Comment:
Yes it always return the result in batch dimension (shape `(1, ...)`). Here's a response sample:
```
INFO:root:{"inference": {"example": "i am happy", "inference": [{"label": "POSITIVE", "score": 0.9998801946640015}], "model_id": "8468081215851724800"}, "video_id": "XpVt6Z1Gjjo", "comment_text": "i am happy", "likes": 1, "replies": 1}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3104317444
Apologies for the super late update on this one... Tests have finally passed now.
Thanks for the review!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207688014
##########
sdks/python/apache_beam/yaml/examples/README.md:
##########
@@ -231,8 +231,9 @@ gcloud dataflow yaml run $JOB_NAME \
### ML
-These examples leverage the built-in `Enrichment` transform for performing
-ML enrichments.
+These examples include the built-in `Enrichment` transform for performing
Review Comment:
Added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207658797
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
+ input: ReadFromMyTopic
+ config:
+ language: python
+ fields:
+ video_id: video_id
+ comment_text:
+ callable: |
+ import re
+ def filter(row):
+ # Match letters, digits, whitespace and common punctuation
+ allowed = r"A-Za-z0-9\s.,;:!?\'\"()\[\]{}\-_"
Review Comment:
See previous comments also. I've expanded the regex match to keep all printable ascii characters instead of manually specifying common punctuations.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3101190666
Please fix the lint failure.
conftest.py gen_protos.py setup.py |
-- | --
Running pylint... |
************* Module apache_beam.yaml.examples.testing.examples_test |
apache_beam/yaml/examples/testing/examples_test.py:863:0: C0301: Line too long (82/80) (line-too-long)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "derrickaw (via GitHub)" <gi...@apache.org>.
derrickaw commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2211169228
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -782,9 +853,67 @@ def _db_io_read_test_processor(
return test_spec
+@YamlExamplesTestSuite.register_test_preprocessor(
+ 'test_streaming_sentiment_analysis_yaml')
+def _streaming_sentiment_analysis_test_preprocessor(
+ test_spec: dict, expected: List[str], env: TestEnvironment):
+ """
+ Preprocessor for tests that involve the streaming sentiment analysis example.
+
+ This preprocessor replaces several IO transforms and the RunInference transform.
+ This allows the test to verify the pipeline's correctness without relying on
+ external data sources and the model hosted on VertexAI.
+
+ Args:
+ test_spec: The dictionary representation of the YAML pipeline specification.
+ expected: A list of strings representing the expected output of the
+ pipeline.
+ env: The TestEnvironment object providing utilities for creating temporary
+ files.
+
+ Returns:
+ The modified test_spec dictionary with ... transforms replaced.
+ """
+ if pipeline := test_spec.get('pipeline', None):
+ for transform in pipeline.get('transforms', []):
+ if transform.get('type', '') == 'PyTransform' and transform.get(
+ 'name', '') == 'ReadFromGCS':
+ transform['windowing'] = {'type': 'fixed', 'size': '30s'}
+
+ file_name = 'youtube-comments.csv'
+ local_path = env.input_file(file_name, INPUT_FILES[file_name])
+ transform['config']['kwargs']['file_pattern'] = local_path
+
+ if pipeline := test_spec.get('pipeline', None):
+ for transform in pipeline.get('transforms', []):
+ if transform.get('type', '') == 'ReadFromKafka':
+ config = transform['config']
+ transform['type'] = 'ReadFromCsv'
+ transform['config'] = {
+ k: v
+ for k, v in config.items() if k.startswith('__')
+ }
+ transform['config']['path'] = ""
+
+ file_name = 'youtube-comments.csv'
+ test_spec = replace_recursive(
+ test_spec,
+ transform['type'],
+ 'path',
+ env.input_file(file_name, INPUT_FILES[file_name]))
+
+ if pipeline := test_spec.get('pipeline', None):
+ for transform in pipeline.get('transforms', []):
+ if transform.get('type', '') == 'RunInference':
+ transform['type'] = 'TestRunInference'
+
Review Comment:
Do we need three if statements or can we combine and have one for loop iteration looping through to check for each type of transform?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3039637166
## [Codecov](https://app.codecov.io/gh/apache/beam/pull/35375?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 56.53%. Comparing base [(`d5eea11`)](https://app.codecov.io/gh/apache/beam/commit/d5eea11b6fb5c7f5fd8f107be19dc713acb2e0ee?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`2666307`)](https://app.codecov.io/gh/apache/beam/commit/2666307e8ef0f607b247bf4b3dc9d5c4d3e1994b?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 3 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #35375 +/- ##
============================================
- Coverage 56.53% 56.53% -0.01%
Complexity 3319 3319
============================================
Files 1199 1199
Lines 183097 183099 +2
Branches 3426 3426
============================================
- Hits 103519 103515 -4
- Misses 76279 76285 +6
Partials 3299 3299
```
| [Flag](https://app.codecov.io/gh/apache/beam/pull/35375/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [python](https://app.codecov.io/gh/apache/beam/pull/35375/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `80.77% <100.00%> (-0.01%)` | :arrow_down: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/beam/pull/35375?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-3090099487
Retest this please
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-2988449715
cc @chamikaramj and @damccorm for feedbacks on whether this example makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207719948
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/README.md:
##########
@@ -0,0 +1,95 @@
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+-->
+
+# Inference Examples Catalog
+
+<!-- TOC -->
+
+* [Inference Examples Catalog](#examples-catalog)
+ * [Streaming Sentiment Analysis](#streaming-sentiment-analysis)
+
+<!-- TOC -->
+
+## Streaming Sentiment Analysis
+
+The example leverages the `RunInference` transform with Vertex AI
+model handler [VertexAIModelHandlerJSON](
+https://beam.apache.org/releases/pydoc/current/apache_beam.yaml.yaml_ml#apache_beam.yaml.yaml_ml.VertexAIModelHandlerJSONProvider),
+in addition to Kafka IO to demonstrate an end-to-end example of a
+streaming sentiment analysis pipeline. The dataset to perform
+sentiment analysis on is the YouTube video comments and can be found
+on Kaggle [here](
+https://www.kaggle.com/datasets/datasnaek/youtube?select=UScomments.csv).
+
+Download the dataset and copy over to a GCS bucket:
+```sh
+gcloud storage cp /path/to/UScomments.csv gs://YOUR_BUCKET/UScomments.csv
+```
+
+For setting up Kafka, an option is to use [Click to Deploy](
Review Comment:
Setting up Kafka with Click to Deploy only takes a few button clicks. This doesn't require any manual setup/installation. I've also added link to our [existing Kafka pipeline example](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/yaml/examples/README.md#io).
As for creating BQ dataset, I've added a command line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207686904
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -114,14 +114,31 @@ def _fn(row):
@beam.ptransform.ptransform_fn
def test_kafka_read(
- pcoll,
+ pbegin,
Review Comment:
Fixed
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -114,14 +114,31 @@ def _fn(row):
@beam.ptransform.ptransform_fn
def test_kafka_read(
- pcoll,
+ pbegin,
format,
topic,
bootstrap_servers,
auto_offset_reset_config,
consumer_config):
+ """
+ This PTransform simulates the behavior of the ReadFromKafka transform
+ with the RAW format by simply using some fixed sample text data and
+ encode it to raw bytes.
+
+ Args:
+ pbegin: The input PCollection.
+ format:
Review Comment:
Added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #35375:
URL: https://github.com/apache/beam/pull/35375#issuecomment-2988667246
Assigning reviewers:
R: @jrmccluskey for label python.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207594378
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
Review Comment:
There are csv rows that look like this:
```
video_id,comment_text,likes,replies
XpVt6Z1Gjjo,I am always,happy0,0
```
I had the helper function `_to_int` to try to convert the `likes` and `replies` columns when reading the csv file, and in this case it converts to `None`. The following `beam.Filter` is to filter out this case. Even if we replace with some default values instead of `None`, I still think it doesn't make sense to keep `broken` rows like this...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2207726224
##########
sdks/python/apache_beam/yaml/examples/transforms/ml/inference/streaming_sentiment_analysis.yaml:
##########
@@ -0,0 +1,257 @@
+# coding=utf-8
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# The pipeline first reads the YouTube comments .csv dataset from GCS bucket
+# and performs necessary clean-up before writing it to a Kafka topic.
+# The pipeline then reads from that Kafka topic and applies various transformation
+# logic before RunInference transform performs remote inference with the Vertex AI
+# model handler.
+# The inference result is then written to a BigQuery table.
+
+pipeline:
+ transforms:
+ # The YouTube comments dataset contains rows that
+ # have unexpected schema (e.g. rows with more fields,
+ # rows with fields that contain string instead of
+ # integer, etc...). PyTransform helps construct
+ # the logic to properly read in the csv dataset as
+ # a schema'd PCollection.
+ - type: PyTransform
+ name: ReadFromGCS
+ input: {}
+ config:
+ constructor: __callable__
+ kwargs:
+ source: |
+ def ReadYoutubeCommentsCsv(pcoll, file_pattern):
+ def _to_int(x):
+ try:
+ return int(x)
+ except (ValueError):
+ return None
+
+ return (
+ pcoll
+ | beam.io.ReadFromCsv(
+ file_pattern,
+ names=['video_id', 'comment_text', 'likes', 'replies'],
+ on_bad_lines='skip',
+ converters={'likes': _to_int, 'replies': _to_int})
+ | beam.Filter(lambda row:
+ None not in list(row._asdict().values()))
+ | beam.Map(lambda row: beam.Row(
+ video_id=row.video_id,
+ comment_text=row.comment_text,
+ likes=int(row.likes),
+ replies=int(row.replies)))
+ )
+ file_pattern: "{{ GCS_PATH }}"
+
+ # Send the rows as Kafka records to an existing
+ # Kafka topic.
+ - type: WriteToKafka
+ name: SendRecordsToKafka
+ input: ReadFromGCS
+ config:
+ format: "JSON"
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ producer_config_updates:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Read Kafka records from an existing Kafka topic.
+ - type: ReadFromKafka
+ name: ReadFromMyTopic
+ config:
+ format: "JSON"
+ schema: |
+ {
+ "type": "object",
+ "properties": {
+ "video_id": { "type": "string" },
+ "comment_text": { "type": "string" },
+ "likes": { "type": "integer" },
+ "replies": { "type": "integer" }
+ }
+ }
+ topic: "{{ TOPIC }}"
+ bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}"
+ auto_offset_reset_config: earliest
+ consumer_config:
+ sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \
+ username={{ USERNAME }} \
+ password={{ PASSWORD }};"
+ security.protocol: "SASL_PLAINTEXT"
+ sasl.mechanism: "PLAIN"
+
+ # Remove unexpected characters from the YouTube
+ # comment string, e.g. emojis, ascii characters outside
+ # the common day-to-day English.
+ - type: MapToFields
+ name: RemoveWeirdCharacters
Review Comment:
The model `distilbert-base-uncased-finetuned-sst-2-english` is trained on the text corpus that doesn't include emojis or non-printable ascii characters [1]. Otherwise we would have false positives, e.g. 😩 or §ÁĐ always give POSITIVE label. I don't think it's meaningful to keep these kind of characters.
[1] https://www.ascii-code.com/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "charlespnh (via GitHub)" <gi...@apache.org>.
charlespnh commented on code in PR #35375:
URL: https://github.com/apache/beam/pull/35375#discussion_r2222725097
##########
sdks/python/apache_beam/yaml/examples/testing/examples_test.py:
##########
@@ -782,9 +853,67 @@ def _db_io_read_test_processor(
return test_spec
+@YamlExamplesTestSuite.register_test_preprocessor(
+ 'test_streaming_sentiment_analysis_yaml')
+def _streaming_sentiment_analysis_test_preprocessor(
+ test_spec: dict, expected: List[str], env: TestEnvironment):
+ """
+ Preprocessor for tests that involve the streaming sentiment analysis example.
+
+ This preprocessor replaces several IO transforms and the RunInference transform.
+ This allows the test to verify the pipeline's correctness without relying on
+ external data sources and the model hosted on VertexAI.
+
+ Args:
+ test_spec: The dictionary representation of the YAML pipeline specification.
+ expected: A list of strings representing the expected output of the
+ pipeline.
+ env: The TestEnvironment object providing utilities for creating temporary
+ files.
+
+ Returns:
+ The modified test_spec dictionary with ... transforms replaced.
+ """
+ if pipeline := test_spec.get('pipeline', None):
+ for transform in pipeline.get('transforms', []):
+ if transform.get('type', '') == 'PyTransform' and transform.get(
+ 'name', '') == 'ReadFromGCS':
+ transform['windowing'] = {'type': 'fixed', 'size': '30s'}
+
+ file_name = 'youtube-comments.csv'
+ local_path = env.input_file(file_name, INPUT_FILES[file_name])
+ transform['config']['kwargs']['file_pattern'] = local_path
+
+ if pipeline := test_spec.get('pipeline', None):
+ for transform in pipeline.get('transforms', []):
+ if transform.get('type', '') == 'ReadFromKafka':
+ config = transform['config']
+ transform['type'] = 'ReadFromCsv'
+ transform['config'] = {
+ k: v
+ for k, v in config.items() if k.startswith('__')
+ }
+ transform['config']['path'] = ""
+
+ file_name = 'youtube-comments.csv'
+ test_spec = replace_recursive(
+ test_spec,
+ transform['type'],
+ 'path',
+ env.input_file(file_name, INPUT_FILES[file_name]))
+
+ if pipeline := test_spec.get('pipeline', None):
+ for transform in pipeline.get('transforms', []):
+ if transform.get('type', '') == 'RunInference':
+ transform['type'] = 'TestRunInference'
+
Review Comment:
Yeah because we're using `replace_recursive` it returns a new `test_spec` reference. You therefore would need to get the new `pipeline` reference.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [YAML] A Streaming Inference Pipeline - YouTube Comments Sentiment Analysis [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
chamikaramj merged PR #35375:
URL: https://github.com/apache/beam/pull/35375
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org