damccorm commented on code in PR #24459:
URL: https://github.com/apache/beam/pull/24459#discussion_r1043589048
##########
playground/infrastructure/helper.py:
##########
@@ -484,29 +522,40 @@ def validate_example_fields(example: Example):
:param example: example from the repository
"""
if example.filepath == "":
- err_msg = f"Example doesn't have a file path field. Example: {example}"
- logging.error(err_msg)
- raise ValidationException(err_msg)
+ _log_and_rise_validation_err(f"Example doesn't have a file path field.
Example: {example}")
if example.name == "":
- err_msg = f"Example doesn't have a name field. Path:
{example.filepath}"
- logging.error(err_msg)
- raise ValidationException(err_msg)
+ _log_and_rise_validation_err(f"Example doesn't have a name field.
Path: {example.filepath}")
if example.sdk == SDK_UNSPECIFIED:
- err_msg = f"Example doesn't have a sdk field. Path: {example.filepath}"
- logging.error(err_msg)
- raise ValidationException(err_msg)
+ _log_and_rise_validation_err(f"Example doesn't have a sdk field. Path:
{example.filepath}")
if example.code == "":
- err_msg = f"Example doesn't have a code field. Path:
{example.filepath}"
- logging.error(err_msg)
- raise ValidationException(err_msg)
+ _log_and_rise_validation_err(f"Example doesn't have a code field.
Path: {example.filepath}")
if example.link == "":
- err_msg = f"Example doesn't have a link field. Path:
{example.filepath}"
- logging.error(err_msg)
- raise ValidationException(err_msg)
+ _log_and_rise_validation_err(f"Example doesn't have a link field.
Path: {example.filepath}")
if example.complexity == "":
- err_msg = f"Example doesn't have a complexity field. Path:
{example.filepath}"
- logging.error(err_msg)
- raise ValidationException(err_msg)
+ _log_and_rise_validation_err(f"Example doesn't have a complexity
field. Path: {example.filepath}")
+ datasets = example.datasets
+ emulators = example.emulators
+
+ if datasets and not emulators:
+ _log_and_rise_validation_err(f"Example has a datasets field but an
emulators field not found. Path: {example.filepath}")
+ if emulators and not datasets:
+ _log_and_rise_validation_err(f"Example has an emulators field but a
datasets field not found. Path: {example.filepath}")
+
+ dataset_names = []
+ for dataset in datasets:
+ location = dataset.location
+ dataset_format = dataset.format
+ if not location or not dataset_format or location not in ["local"] or
dataset_format not in ["json", "avro"]:
+ _log_and_rise_validation_err(f"Example has invalid dataset value.
Path: {example.filepath}")
+ dataset_names.append(dataset.name)
+ for emulator in emulators:
+ if not (emulator.name == "kafka" and emulator.topic.dataset in
dataset_names):
+ _log_and_rise_validation_err(f"Example has invalid emulator value.
Path: {example.filepath}")
+
+
+def _log_and_rise_validation_err(msg: str):
Review Comment:
```suggestion
def _log_and_raise_validation_err(msg: str):
```
Nit: this will need to be updated on the other lines as well.
##########
playground/backend/internal/db/mapper/precompiled_object_mapper_test.go:
##########
@@ -41,7 +41,11 @@ func TestPrecompiledObjectMapper_ToObjectInfo(t *testing.T) {
actualResult.ContextLine != 32 ||
len(actualResult.Categories) != 3 ||
actualResult.Type.String() != "PRECOMPILED_OBJECT_TYPE_EXAMPLE"
||
- actualResult.Sdk != pb.Sdk_SDK_JAVA {
+ actualResult.Sdk != pb.Sdk_SDK_JAVA ||
+ len(actualResult.Datasets) != 1 ||
+ actualResult.Datasets[0].DatasetPath != "MOCK_PATH_0" ||
+ actualResult.Datasets[0].Options["Topic"] != "MOCK_TOPIC" ||
+ actualResult.Datasets[0].Type !=
pb.EmulatorType_EMULATOR_TYPE_KAFKA {
t.Error("ToObjectInfo() unexpected result")
Review Comment:
Could we split these into separate assertions with informative error
messages? Same comment below
##########
examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountAvro.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+package org.apache.beam.examples;
+
+// beam-playground:
+// name: KafkaWordCountAvro
+// description: Test example with Apache Kafka
+// multifile: false
+// context_line: 55
+// categories:
+// - Filtering
+// - Options
+// - Quickstart
+// complexity: MEDIUM
+// tags:
+// - filter
+// - strings
+// - emulator
+// emulators:
+// kafka:
+// topic:
+// id: dataset
+// dataset: dataset
+// datasets:
+// dataset:
+// location: local
+// format: avro
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.SimpleFunction;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.values.KV;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+public class KafkaWordCountAvro {
+ static final String TOKENIZER_PATTERN = "[^\\p{L}]+";
Review Comment:
Could you add a comment explaining what this is doing? Why are these our
tokenizing characters? Same comment applies below
##########
playground/infrastructure/datastore_client.py:
##########
@@ -83,54 +85,65 @@ def save_to_cloud_datastore(self, examples_from_rep:
List[Example], sdk: Sdk, or
examples_ids_before_updating = self._get_all_examples(sdk, origin)
# loop through every example to save them to the Cloud Datastore
Review Comment:
Would we be better off splitting this up into multiple transactions? What
would the consequences of this non-transaction getting interrupted?
##########
examples/java/src/main/java/org/apache/beam/examples/KafkaWordCountAvro.java:
##########
@@ -0,0 +1,118 @@
+/*
+ * 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.
+ */
+package org.apache.beam.examples;
+
+// beam-playground:
+// name: KafkaWordCountAvro
+// description: Test example with Apache Kafka
+// multifile: false
+// context_line: 55
+// categories:
+// - Filtering
+// - Options
+// - Quickstart
+// complexity: MEDIUM
+// tags:
+// - filter
+// - strings
+// - emulator
+// emulators:
+// kafka:
+// topic:
+// id: dataset
+// dataset: dataset
+// datasets:
+// dataset:
+// location: local
+// format: avro
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.io.TextIO;
+import org.apache.beam.sdk.io.kafka.KafkaIO;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.transforms.Count;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.MapElements;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.SimpleFunction;
+import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.values.KV;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.LongDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
+
+public class KafkaWordCountAvro {
+ static final String TOKENIZER_PATTERN = "[^\\p{L}]+";
+
+ public static void main(String[] args) {
+ final PipelineOptions options = PipelineOptionsFactory.create();
+ final Pipeline p = Pipeline.create(options);
+
+ final Map<String, Object> consumerConfig = new HashMap<>();
+ consumerConfig.put("auto.offset.reset", "earliest");
+
+ p.apply(
+ KafkaIO.<Long, String>read()
+ .withBootstrapServers(
+ "kafka_server:9092") // The argument is predefined to a
correct value. Do not
Review Comment:
Where does this value come from? Could you add a little more context in the
comment? (Same comment applies to withTopicPartitions argument below)
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
