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-unsubscr...@beam.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
