From 7b10f5b3edd890714b9f87fbfa80a1d04c2c47ee Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Sun, 4 Oct 2026 12:24:42 +0000 Subject: [PATCH 1/6] kafka edits --- yaml/README_Kafka_To_Spanner_Yaml.md | 267 ++++++++++++++++++ .../templates/yaml/KafkaToBigQueryYaml.java | 76 ++--- .../templates/yaml/KafkaToIcebergYaml.java | 99 +++---- .../templates/yaml/KafkaToSpannerYaml.java | 188 ++++++++++++ .../main/python/options/kafka_options.yaml | 50 +--- .../main/python/options/spanner_options.yaml | 31 ++ yaml/src/main/yaml/KafkaToSpanner.yaml | 66 +++++ .../templates/yaml/KafkaToSpannerYamlIT.java | 183 ++++++++++++ 8 files changed, 802 insertions(+), 158 deletions(-) create mode 100644 yaml/README_Kafka_To_Spanner_Yaml.md create mode 100644 yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java create mode 100644 yaml/src/main/python/options/spanner_options.yaml create mode 100644 yaml/src/main/yaml/KafkaToSpanner.yaml create mode 100644 yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java diff --git a/yaml/README_Kafka_To_Spanner_Yaml.md b/yaml/README_Kafka_To_Spanner_Yaml.md new file mode 100644 index 0000000000..1517bee210 --- /dev/null +++ b/yaml/README_Kafka_To_Spanner_Yaml.md @@ -0,0 +1,267 @@ + +Kafka to Spanner (YAML) template +--- +The Kafka to Spanner template is a streaming pipeline which ingests data from an +Apache Kafka topic and writes the resulting records to Cloud Spanner. + + +:memo: This is a Google-provided template! Please +check [Provided templates documentation](https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-spanner) +on how to use it without having to build from sources using [Create job from template](https://console.cloud.google.com/dataflow/createjob?template=Kafka_To_Spanner_Yaml). + +:bulb: This is a generated documentation based +on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/contributor-docs/code-contributions.md#metadata-annotations) +. Do not change this file directly. + +## Parameters + +### Required parameters + +* **bootstrapServers**: A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. For example: host1:port1,host2:port2 For example, `host1:port1,host2:port2,localhost:9092,127.0.0.1:9093`. +* **topic**: Kafka topic to read from. For example: my_topic For example, `my_topic`. +* **instanceId**: The Cloud Spanner instance ID. For example, `your-instance-id`. +* **databaseId**: The Cloud Spanner database ID. For example, `your-database-id`. +* **tableId**: The Cloud Spanner table ID to write the output to. For example, `your-table-id`. + +### Optional parameters + +* **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. +* **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. +* **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. +* **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. +* **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. +* **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. +* **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. +* **projectId**: The Google Cloud project ID of the Cloud Spanner instance. For example, `your-project-id`. + + + +## Getting Started + +### Requirements + +* Java 17 +* Maven +* [gcloud CLI](https://cloud.google.com/sdk/gcloud), and execution of the + following commands: + * `gcloud auth login` + * `gcloud auth application-default login` + +:star2: Those dependencies are pre-installed if you use Google Cloud Shell! + +[![Open in Cloud Shell](http://gstatic.com/cloudssh/images/open-btn.svg)](https://console.cloud.google.com/cloudshell/editor?cloudshell_git_repo=https%3A%2F%2Fgithub.com%2FGoogleCloudPlatform%2FDataflowTemplates.git&cloudshell_open_in_editor=yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java) + +### Templates Plugin + +This README provides instructions using +the [Templates Plugin](https://github.com/GoogleCloudPlatform/DataflowTemplates/blob/main/contributor-docs/code-contributions.md#templates-plugin). + +#### Validating the Template + +This template has a validation command that is used to check code quality. + +```shell +mvn clean install -PtemplatesValidate \ +-DskipTests -am \ +-pl yaml +``` + +### Building Template + +This template is a Flex Template, meaning that the pipeline code will be +containerized and the container will be executed on Dataflow. Please +check [Use Flex Templates](https://cloud.google.com/dataflow/docs/guides/templates/using-flex-templates) +and [Configure Flex Templates](https://cloud.google.com/dataflow/docs/guides/templates/configuring-flex-templates) +for more information. + +#### Staging the Template + +If the plan is to just stage the template (i.e., make it available to use) by +the `gcloud` command or Dataflow "Create job from template" UI, +the `-PtemplatesStage` profile should be used: + +```shell +export PROJECT= +export BUCKET_NAME= +export ARTIFACT_REGISTRY_REPO=-docker.pkg.dev/$PROJECT/ + +mvn clean package -PtemplatesStage \ +-DskipTests \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-DartifactRegistry="$ARTIFACT_REGISTRY_REPO" \ +-DstagePrefix="templates" \ +-DtemplateName="Kafka_To_Spanner_Yaml" \ +-f yaml +``` + +The `-DartifactRegistry` parameter can be specified to set the artifact registry repository of the Flex Templates image. +If not provided, it defaults to `gcr.io/`. + +The command should build and save the template to Google Cloud, and then print +the complete location on Cloud Storage: + +``` +Flex Template was staged! gs:///templates/flex/Kafka_To_Spanner_Yaml +``` + +The specific path should be copied as it will be used in the following steps. + +#### Running the Template + +**Using the staged template**: + +You can use the path above run the template (or share with others for execution). + +To start a job with the template at any time using `gcloud`, you are going to +need valid resources for the required parameters. + +Provided that, the following command line can be used: + +```shell +export PROJECT= +export BUCKET_NAME= +export REGION=us-central1 +export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/Kafka_To_Spanner_Yaml" + +### Required +export BOOTSTRAP_SERVERS= +export TOPIC= +export INSTANCE_ID= +export DATABASE_ID= +export TABLE_ID= + +### Optional +export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= +export CONFLUENT_SCHEMA_REGISTRY_URL= +export CONSUMER_CONFIG_UPDATES= +export FILE_DESCRIPTOR_PATH= +export FORMAT=JSON +export MAX_READ_TIME_SECONDS= +export MESSAGE_NAME= +export SCHEMA= +export PROJECT_ID= + +gcloud dataflow flex-template run "kafka-to-spanner-yaml-job" \ + --project "$PROJECT" \ + --region "$REGION" \ + --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ + --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ + --parameters "topic=$TOPIC" \ + --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ + --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ + --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ + --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ + --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ + --parameters "messageName=$MESSAGE_NAME" \ + --parameters "schema=$SCHEMA" \ + --parameters "projectId=$PROJECT_ID" \ + --parameters "instanceId=$INSTANCE_ID" \ + --parameters "databaseId=$DATABASE_ID" \ + --parameters "tableId=$TABLE_ID" +``` + +For more information about the command, please check: +https://cloud.google.com/sdk/gcloud/reference/dataflow/flex-template/run + + +**Using the plugin**: + +Instead of just generating the template in the folder, it is possible to stage +and run the template in a single command. This may be useful for testing when +changing the templates. + +```shell +export PROJECT= +export BUCKET_NAME= +export REGION=us-central1 + +### Required +export BOOTSTRAP_SERVERS= +export TOPIC= +export INSTANCE_ID= +export DATABASE_ID= +export TABLE_ID= + +### Optional +export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= +export CONFLUENT_SCHEMA_REGISTRY_URL= +export CONSUMER_CONFIG_UPDATES= +export FILE_DESCRIPTOR_PATH= +export FORMAT=JSON +export MAX_READ_TIME_SECONDS= +export MESSAGE_NAME= +export SCHEMA= +export PROJECT_ID= + +mvn clean package -PtemplatesRun \ +-DskipTests \ +-DprojectId="$PROJECT" \ +-DbucketName="$BUCKET_NAME" \ +-Dregion="$REGION" \ +-DjobName="kafka-to-spanner-yaml-job" \ +-DtemplateName="Kafka_To_Spanner_Yaml" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,databaseId=$DATABASE_ID,tableId=$TABLE_ID" \ +-f yaml +``` + +## Terraform + +Dataflow supports the utilization of Terraform to manage template jobs, +see [dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job). + +Terraform modules have been generated for most templates in this repository. This includes the relevant parameters +specific to the template. If available, they may be used instead of +[dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job) +directly. + +To use the autogenerated module, execute the standard +[terraform workflow](https://developer.hashicorp.com/terraform/intro/core-workflow): + +```shell +cd yaml/terraform/Kafka_To_Spanner_Yaml +terraform init +terraform apply +``` + +To use +[dataflow_flex_template_job](https://registry.terraform.io/providers/hashicorp/google/latest/docs/resources/dataflow_flex_template_job) +directly: + +```terraform +provider "google-beta" { + project = var.project +} +variable "project" { + default = "" +} +variable "region" { + default = "us-central1" +} + +resource "google_dataflow_flex_template_job" "kafka_to_spanner_yaml" { + + provider = google-beta + container_spec_gcs_path = "gs://dataflow-templates-${var.region}/latest/flex/Kafka_To_Spanner_Yaml" + name = "kafka-to-spanner-yaml" + region = var.region + parameters = { + bootstrapServers = "" + topic = "" + instanceId = "" + databaseId = "" + tableId = "" + # confluentSchemaRegistrySubject = "" + # confluentSchemaRegistryUrl = "" + # consumerConfigUpdates = "" + # fileDescriptorPath = "" + # format = "JSON" + # maxReadTimeSeconds = "" + # messageName = "" + # schema = "" + # projectId = "" + } +} +``` diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java index 56e9c2256b..f9ccee7f91 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java @@ -65,17 +65,8 @@ public interface KafkaToBigQueryYaml { @Validation.Required String getTopic(); - @TemplateParameter.Boolean( - order = 3, - name = "allowDuplicates", - optional = true, - description = "If the Kafka read allows duplicates.", - helpText = "If the Kafka read allows duplicates. For example: true", - example = "true") - Boolean getAllowDuplicates(); - @TemplateParameter.Text( - order = 4, + order = 3, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -84,7 +75,7 @@ public interface KafkaToBigQueryYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 5, + order = 4, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -94,7 +85,7 @@ public interface KafkaToBigQueryYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 6, + order = 5, name = "consumerConfigUpdates", optional = true, description = @@ -105,7 +96,7 @@ public interface KafkaToBigQueryYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 7, + order = 6, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -115,7 +106,7 @@ public interface KafkaToBigQueryYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 8, + order = 7, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -125,6 +116,15 @@ public interface KafkaToBigQueryYaml { @Default.String("JSON") String getFormat(); + @TemplateParameter.Integer( + order = 8, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + @TemplateParameter.Text( order = 9, name = "messageName", @@ -136,44 +136,8 @@ public interface KafkaToBigQueryYaml { example = "MyMessage") String getMessageName(); - @TemplateParameter.Boolean( - order = 10, - name = "offsetDeduplication", - optional = true, - description = "If the redistribute is using offset deduplication mode.", - helpText = "If the redistribute is using offset deduplication mode. For example: true", - example = "true") - Boolean getOffsetDeduplication(); - - @TemplateParameter.Boolean( - order = 11, - name = "redistributeByRecordKey", - optional = true, - description = "If the redistribute keys by the Kafka record key.", - helpText = "If the redistribute keys by the Kafka record key. For example: true", - example = "true") - Boolean getRedistributeByRecordKey(); - - @TemplateParameter.Integer( - order = 12, - name = "redistributeNumKeys", - optional = true, - description = "The number of keys for redistributing Kafka inputs.", - helpText = "The number of keys for redistributing Kafka inputs. For example: 10", - example = "10") - Integer getRedistributeNumKeys(); - - @TemplateParameter.Boolean( - order = 13, - name = "redistributed", - optional = true, - description = "If the Kafka read should be redistributed.", - helpText = "If the Kafka read should be redistributed. For example: true", - example = "true") - Boolean getRedistributed(); - @TemplateParameter.Text( - order = 14, + order = 10, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -184,7 +148,7 @@ public interface KafkaToBigQueryYaml { String getSchema(); @TemplateParameter.Text( - order = 15, + order = 11, name = "table", optional = false, description = "BigQuery table", @@ -195,7 +159,7 @@ public interface KafkaToBigQueryYaml { String getTable(); @TemplateParameter.Text( - order = 16, + order = 12, name = "createDisposition", optional = true, description = "How to create", @@ -206,7 +170,7 @@ public interface KafkaToBigQueryYaml { String getCreateDisposition(); @TemplateParameter.Text( - order = 17, + order = 13, name = "writeDisposition", optional = true, description = "How to write", @@ -217,7 +181,7 @@ public interface KafkaToBigQueryYaml { String getWriteDisposition(); @TemplateParameter.Integer( - order = 18, + order = 14, name = "numStreams", optional = true, description = "Number of streams for BigQuery Storage Write API", @@ -228,7 +192,7 @@ public interface KafkaToBigQueryYaml { Integer getNumStreams(); @TemplateParameter.Text( - order = 19, + order = 15, name = "outputDeadletterTable", optional = false, description = "The dead-letter table name to output failed messages to BigQuery", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java index 7ddc2e9f59..072650202c 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java @@ -68,17 +68,8 @@ public interface KafkaToIcebergYaml { @Validation.Required String getTopic(); - @TemplateParameter.Boolean( - order = 3, - name = "allowDuplicates", - optional = true, - description = "If the Kafka read allows duplicates.", - helpText = "If the Kafka read allows duplicates. For example: true", - example = "true") - Boolean getAllowDuplicates(); - @TemplateParameter.Text( - order = 4, + order = 3, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -87,7 +78,7 @@ public interface KafkaToIcebergYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 5, + order = 4, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -97,7 +88,7 @@ public interface KafkaToIcebergYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 6, + order = 5, name = "consumerConfigUpdates", optional = true, description = @@ -108,7 +99,7 @@ public interface KafkaToIcebergYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 7, + order = 6, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -118,7 +109,7 @@ public interface KafkaToIcebergYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 8, + order = 7, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -128,6 +119,15 @@ public interface KafkaToIcebergYaml { @Default.String("JSON") String getFormat(); + @TemplateParameter.Integer( + order = 8, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + @TemplateParameter.Text( order = 9, name = "messageName", @@ -139,44 +139,8 @@ public interface KafkaToIcebergYaml { example = "MyMessage") String getMessageName(); - @TemplateParameter.Boolean( - order = 10, - name = "offsetDeduplication", - optional = true, - description = "If the redistribute is using offset deduplication mode.", - helpText = "If the redistribute is using offset deduplication mode. For example: true", - example = "true") - Boolean getOffsetDeduplication(); - - @TemplateParameter.Boolean( - order = 11, - name = "redistributeByRecordKey", - optional = true, - description = "If the redistribute keys by the Kafka record key.", - helpText = "If the redistribute keys by the Kafka record key. For example: true", - example = "true") - Boolean getRedistributeByRecordKey(); - - @TemplateParameter.Integer( - order = 12, - name = "redistributeNumKeys", - optional = true, - description = "The number of keys for redistributing Kafka inputs.", - helpText = "The number of keys for redistributing Kafka inputs. For example: 10", - example = "10") - Integer getRedistributeNumKeys(); - - @TemplateParameter.Boolean( - order = 13, - name = "redistributed", - optional = true, - description = "If the Kafka read should be redistributed.", - helpText = "If the Kafka read should be redistributed. For example: true", - example = "true") - Boolean getRedistributed(); - @TemplateParameter.Text( - order = 14, + order = 10, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -187,7 +151,7 @@ public interface KafkaToIcebergYaml { String getSchema(); @TemplateParameter.Text( - order = 15, + order = 11, name = "table", optional = false, description = "A fully-qualified table identifier.", @@ -197,7 +161,7 @@ public interface KafkaToIcebergYaml { String getTable(); @TemplateParameter.Text( - order = 16, + order = 12, name = "catalogName", optional = false, description = "Name of the catalog containing the table.", @@ -207,7 +171,7 @@ public interface KafkaToIcebergYaml { String getCatalogName(); @TemplateParameter.Text( - order = 17, + order = 13, name = "catalogProperties", optional = false, description = "Properties used to set up the Iceberg catalog.", @@ -217,7 +181,7 @@ public interface KafkaToIcebergYaml { String getCatalogProperties(); @TemplateParameter.Text( - order = 18, + order = 14, name = "configProperties", optional = true, description = "Properties passed to the Hadoop Configuration.", @@ -226,7 +190,7 @@ public interface KafkaToIcebergYaml { String getConfigProperties(); @TemplateParameter.Text( - order = 19, + order = 15, name = "drop", optional = true, description = "A list of field names to drop from the input record before writing.", @@ -235,7 +199,16 @@ public interface KafkaToIcebergYaml { String getDrop(); @TemplateParameter.Text( - order = 20, + order = 16, + name = "filter", + optional = true, + description = "An optional filter expression to apply to the input records.", + helpText = "A filter expression to apply to records from the Iceberg table.", + example = "age > 18") + String getFilter(); + + @TemplateParameter.Text( + order = 17, name = "keep", optional = true, description = "A list of field names to keep in the input record.", @@ -244,7 +217,7 @@ public interface KafkaToIcebergYaml { String getKeep(); @TemplateParameter.Text( - order = 21, + order = 18, name = "only", optional = true, description = "The name of a single record field that should be written.", @@ -253,7 +226,7 @@ public interface KafkaToIcebergYaml { String getOnly(); @TemplateParameter.Text( - order = 22, + order = 19, name = "partitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", @@ -262,7 +235,7 @@ public interface KafkaToIcebergYaml { String getPartitionFields(); @TemplateParameter.Text( - order = 23, + order = 20, name = "tableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", @@ -271,7 +244,7 @@ public interface KafkaToIcebergYaml { String getTableProperties(); @TemplateParameter.Integer( - order = 24, + order = 21, name = "triggeringFrequencySeconds", optional = false, description = "For a streaming pipeline, the frequency at which snapshots are produced.", @@ -281,7 +254,7 @@ public interface KafkaToIcebergYaml { Integer getTriggeringFrequencySeconds(); @TemplateParameter.Text( - order = 25, + order = 22, name = "sdfCheckpointAfterDuration", optional = true, description = "Dataflow Pipeline Option: Duration after which to checkpoint stateful DoFns.", @@ -292,7 +265,7 @@ public interface KafkaToIcebergYaml { String getSdfCheckpointAfterDuration(); @TemplateParameter.Integer( - order = 26, + order = 23, name = "sdfCheckpointAfterOutputBytes", optional = true, description = diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java new file mode 100644 index 0000000000..3ff1404866 --- /dev/null +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java @@ -0,0 +1,188 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.templates.yaml; + +import com.google.cloud.teleport.metadata.Template; +import com.google.cloud.teleport.metadata.TemplateCategory; +import com.google.cloud.teleport.metadata.TemplateParameter; +import org.apache.beam.sdk.options.Default; +import org.apache.beam.sdk.options.Validation; + +@Template( + name = "Kafka_To_Spanner_Yaml", + category = TemplateCategory.STREAMING, + type = Template.TemplateType.YAML, + displayName = "Kafka to Spanner (YAML)", + description = + "The Kafka to Spanner template is a streaming pipeline which ingests data from an Apache Kafka topic and writes the resulting records to Cloud Spanner.", + flexContainerName = "pipeline-yaml", + yamlTemplateFile = "KafkaToSpanner.yaml", + filesToCopy = {"main.py", "requirements.txt"}, + documentation = + "https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-spanner", + contactInformation = "https://cloud.google.com/support", + requirements = { + "The input Apache Kafka topic must exist.", + "The Apache Kafka broker server must be running and be reachable from the Dataflow worker machines.", + "The output Cloud Spanner instance, database, and table must exist." + }, + streaming = true, + hidden = false) +public interface KafkaToSpannerYaml { + + @TemplateParameter.Text( + order = 1, + name = "bootstrapServers", + optional = false, + description = + "A list of host/port pairs to use for establishing the initial connection to the Kafka cluster.", + helpText = + "A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. For example: host1:port1,host2:port2", + example = "host1:port1,host2:port2,localhost:9092,127.0.0.1:9093") + @Validation.Required + String getBootstrapServers(); + + @TemplateParameter.Text( + order = 2, + name = "topic", + optional = false, + description = "Kafka topic to read from.", + helpText = "Kafka topic to read from. For example: my_topic", + example = "my_topic") + @Validation.Required + String getTopic(); + + @TemplateParameter.Text( + order = 3, + name = "confluentSchemaRegistrySubject", + optional = true, + description = "The subject name for the Confluent Schema Registry.", + helpText = "The subject name for the Confluent Schema Registry. For example: my_subject", + example = "my_subject") + String getConfluentSchemaRegistrySubject(); + + @TemplateParameter.Text( + order = 4, + name = "confluentSchemaRegistryUrl", + optional = true, + description = "The URL for the Confluent Schema Registry.", + helpText = + "The URL for the Confluent Schema Registry. For example: http://schema-registry:8081", + example = "http://schema-registry:8081") + String getConfluentSchemaRegistryUrl(); + + @TemplateParameter.Text( + order = 5, + name = "consumerConfigUpdates", + optional = true, + description = + "A list of key-value pairs that act as configuration parameters for Kafka consumers.", + helpText = + "A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'}", + example = "{\"group.id\": \"my_group\"}") + String getConsumerConfigUpdates(); + + @TemplateParameter.Text( + order = 6, + name = "fileDescriptorPath", + optional = true, + description = "The path to the Protocol Buffer File Descriptor Set file.", + helpText = + "The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb", + example = "gs://bucket/path/to/descriptor.pb") + String getFileDescriptorPath(); + + @TemplateParameter.Text( + order = 7, + name = "format", + optional = true, + description = "The encoding format for the data stored in Kafka.", + helpText = + "The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON", + example = "JSON") + @Default.String("JSON") + String getFormat(); + + @TemplateParameter.Integer( + order = 8, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + + @TemplateParameter.Text( + order = 9, + name = "messageName", + optional = true, + description = + "The name of the Protocol Buffer message to be used for schema extraction and data conversion.", + helpText = + "The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage", + example = "MyMessage") + String getMessageName(); + + @TemplateParameter.Text( + order = 10, + name = "schema", + optional = true, + description = "The schema in which the data is encoded in the Kafka topic.", + helpText = + "The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO.", + example = + "{\"type\": \"record\", \"name\": \"User\", \"fields\": [{\"name\": \"name\", \"type\": \"string\"}]}") + String getSchema(); + + @TemplateParameter.Text( + order = 11, + name = "projectId", + optional = true, + description = "Cloud Spanner Project ID.", + helpText = "The Google Cloud project ID of the Cloud Spanner instance.", + example = "your-project-id") + String getProjectId(); + + @TemplateParameter.Text( + order = 12, + name = "instanceId", + optional = false, + description = "Cloud Spanner Instance ID.", + helpText = "The Cloud Spanner instance ID.", + example = "your-instance-id") + @Validation.Required + String getInstanceId(); + + @TemplateParameter.Text( + order = 13, + name = "databaseId", + optional = false, + description = "Cloud Spanner Database ID.", + helpText = "The Cloud Spanner database ID.", + example = "your-database-id") + @Validation.Required + String getDatabaseId(); + + @TemplateParameter.Text( + order = 14, + name = "tableId", + optional = false, + description = "Cloud Spanner Table ID.", + helpText = "The Cloud Spanner table ID to write the output to.", + example = "your-table-id") + @Validation.Required + String getTableId(); +} diff --git a/yaml/src/main/python/options/kafka_options.yaml b/yaml/src/main/python/options/kafka_options.yaml index d64c4c4e49..2f21c53f87 100644 --- a/yaml/src/main/python/options/kafka_options.yaml +++ b/yaml/src/main/python/options/kafka_options.yaml @@ -16,41 +16,34 @@ options: required: true type: text - order: 3 - name: "allowDuplicates" - description: "If the Kafka read allows duplicates." - help: "If the Kafka read allows duplicates. For example: true" - example: "true" - required: false - type: boolean - - order: 4 name: "confluentSchemaRegistrySubject" description: "The subject name for the Confluent Schema Registry." help: "The subject name for the Confluent Schema Registry. For example: my_subject" example: "my_subject" required: false type: text - - order: 5 + - order: 4 name: "confluentSchemaRegistryUrl" description: "The URL for the Confluent Schema Registry." help: "The URL for the Confluent Schema Registry. For example: http://schema-registry:8081" example: "http://schema-registry:8081" required: false type: text - - order: 6 + - order: 5 name: "consumerConfigUpdates" description: "A list of key-value pairs that act as configuration parameters for Kafka consumers." help: "A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'}" example: '{"group.id": "my_group"}' required: false type: map - - order: 7 + - order: 6 name: "fileDescriptorPath" description: "The path to the Protocol Buffer File Descriptor Set file." help: "The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb" example: "gs://bucket/path/to/descriptor.pb" required: false type: text - - order: 8 + - order: 7 name: "format" description: "The encoding format for the data stored in Kafka." help: "The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON" @@ -58,6 +51,13 @@ options: required: false default: "JSON" type: text + - order: 8 + name: "maxReadTimeSeconds" + description: "Upper bound of how long to read from Kafka." + help: "Upper bound of how long to read from Kafka in seconds. For example: 60" + example: "60" + required: false + type: integer - order: 9 name: "messageName" description: "The name of the Protocol Buffer message to be used for schema extraction and data conversion." @@ -66,34 +66,6 @@ options: required: false type: text - order: 10 - name: "offsetDeduplication" - description: "If the redistribute is using offset deduplication mode." - help: "If the redistribute is using offset deduplication mode. For example: true" - example: "true" - required: false - type: boolean - - order: 11 - name: "redistributeByRecordKey" - description: "If the redistribute keys by the Kafka record key." - help: "If the redistribute keys by the Kafka record key. For example: true" - example: "true" - required: false - type: boolean - - order: 12 - name: "redistributeNumKeys" - description: "The number of keys for redistributing Kafka inputs." - help: "The number of keys for redistributing Kafka inputs. For example: 10" - example: "10" - required: false - type: integer - - order: 13 - name: "redistributed" - description: "If the Kafka read should be redistributed." - help: "If the Kafka read should be redistributed. For example: true" - example: "true" - required: false - type: boolean - - order: 14 name: "schema" description: "The schema in which the data is encoded in the Kafka topic." help: > diff --git a/yaml/src/main/python/options/spanner_options.yaml b/yaml/src/main/python/options/spanner_options.yaml new file mode 100644 index 0000000000..916ab3f633 --- /dev/null +++ b/yaml/src/main/python/options/spanner_options.yaml @@ -0,0 +1,31 @@ +options: + - name: "spanner_common_options" + parameters: + - order: 1 + name: "projectId" + description: "Cloud Spanner Project ID." + help: "The Google Cloud project ID of the Cloud Spanner instance." + example: "your-project-id" + required: false + type: text + - order: 2 + name: "instanceId" + description: "Cloud Spanner Instance ID." + help: "The Cloud Spanner instance ID." + example: "your-instance-id" + required: true + type: text + - order: 3 + name: "databaseId" + description: "Cloud Spanner Database ID." + help: "The Cloud Spanner database ID." + example: "your-database-id" + required: true + type: text + - order: 4 + name: "tableId" + description: "Cloud Spanner Table ID." + help: "The Cloud Spanner table ID to write the output to." + example: "your-table-id" + required: true + type: text diff --git a/yaml/src/main/yaml/KafkaToSpanner.yaml b/yaml/src/main/yaml/KafkaToSpanner.yaml new file mode 100644 index 0000000000..e300303bfd --- /dev/null +++ b/yaml/src/main/yaml/KafkaToSpanner.yaml @@ -0,0 +1,66 @@ +template: + + name: "Kafka_To_Spanner_Yaml" + category: "STREAMING" + type: "YAML" + display_name: "Kafka to Spanner (YAML)" + description: > + The Kafka to Spanner template is a streaming pipeline which ingests + data from an Apache Kafka topic and writes the resulting records to + Cloud Spanner. + flex_container_name: "pipeline-yaml" + yamlTemplateFile: "KafkaToSpanner.yaml" + filesToCopy: > + {"main.py", "requirements.txt"} + documentation: > + https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-spanner + contactInformation: "https://cloud.google.com/support" + requirements: { + "The input Apache Kafka topic must exist.", + "The Apache Kafka broker server must be running and be reachable from the Dataflow worker machines.", + "The output Cloud Spanner instance, database, and table must exist." + } + streaming: true + hidden: false + + options_file: + - "kafka_options" + - "spanner_options" + + parameters: + - kafka_read_options + - spanner_common_options + +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: ReadFromKafka + config: + bootstrap_servers: {{ bootstrapServers }} + topic: {{ topic }} + confluent_schema_registry_subject: {{ confluentSchemaRegistrySubject }} + confluent_schema_registry_url: {{ confluentSchemaRegistryUrl }} + consumer_config: {{ consumerConfigUpdates }} + file_descriptor_path: {{ fileDescriptorPath }} + format: {{ format | default('JSON') }} + max_read_time_seconds: {{ maxReadTimeSeconds }} + message_name: {{ messageName }} + {% if schema %} + schema: | + {{ schema }} + {% endif %} + auto_offset_reset_config: 'earliest' + - type: WriteToSpanner + name: WriteToSpanner + config: + project_id: {{ projectId }} + instance_id: {{ instanceId }} + database_id: {{ databaseId }} + table: {{ tableId }} + windowing: + type: fixed + size: 60 + +options: + streaming: true diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java new file mode 100644 index 0000000000..6a5e23640e --- /dev/null +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYamlIT.java @@ -0,0 +1,183 @@ +/* + * Copyright (C) 2026 Google LLC + * + * Licensed 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 com.google.cloud.teleport.templates.yaml; + +import static com.google.common.truth.Truth.assertThat; +import static org.apache.beam.it.gcp.spanner.matchers.SpannerAsserts.assertThatStructs; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; + +import com.google.cloud.spanner.Struct; +import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import org.apache.beam.it.common.PipelineLauncher; +import org.apache.beam.it.common.PipelineOperator; +import org.apache.beam.it.common.TestProperties; +import org.apache.beam.it.common.utils.ResourceManagerUtils; +import org.apache.beam.it.gcp.TemplateTestBase; +import org.apache.beam.it.gcp.spanner.SpannerResourceManager; +import org.apache.beam.it.gcp.spanner.conditions.SpannerRowsCheck; +import org.apache.beam.it.kafka.KafkaResourceManager; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; +import org.apache.kafka.common.serialization.StringSerializer; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** Integration test for {@link KafkaToSpannerYaml}. */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@TemplateIntegrationTest(KafkaToSpannerYaml.class) +@RunWith(JUnit4.class) +public final class KafkaToSpannerYamlIT extends TemplateTestBase { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaToSpannerYamlIT.class); + + private KafkaResourceManager kafkaResourceManager; + private SpannerResourceManager spannerResourceManager; + + @Before + public void setUp() throws IOException { + kafkaResourceManager = + KafkaResourceManager.builder(testName).setHost(TestProperties.hostIp()).build(); + spannerResourceManager = + SpannerResourceManager.builder(testName, PROJECT, REGION) + .maybeUseStaticInstance() + .setCredentials(credentials) + .build(); + } + + @After + public void tearDown() { + ResourceManagerUtils.cleanResources(kafkaResourceManager, spannerResourceManager); + } + + @Test + public void testKafkaToSpanner() throws IOException { + kafkaToSpanner(Function.identity()); + } + + public void kafkaToSpanner( + Function + paramsAdder) + throws IOException { + + LOG.info("Starting kafkaToSpanner test. Test name: {}. Spec path: {}", testName, specPath); + + // Arrange + LOG.info("Creating Kafka topic..."); + String topicName = kafkaResourceManager.createTopic(testName, 5); + + LOG.info("Creating Spanner table..."); + String tableId = testName; + spannerResourceManager.executeDdlStatement( + String.format( + "CREATE TABLE `%s` (\n" + + " id INT64 NOT NULL,\n" + + " name STRING(1024)\n" + + ") PRIMARY KEY (id)", + tableId)); + + String bootstrapServers = + kafkaResourceManager.getBootstrapServers().replace("PLAINTEXT://", ""); + LOG.info("Using Kafka Bootstrap Servers: {}", bootstrapServers); + + LOG.info("Creating launch config with yaml pipeline parameters..."); + PipelineLauncher.LaunchConfig.Builder options = + paramsAdder.apply( + PipelineLauncher.LaunchConfig.builder(testName, specPath) + .addParameter("bootstrapServers", bootstrapServers) + .addParameter("topic", topicName) + .addParameter("format", "JSON") + .addParameter( + "schema", + "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"integer\"},\"name\":{\"type\":\"string\"}}}") + .addParameter("projectId", PROJECT) + .addParameter("instanceId", spannerResourceManager.getInstanceId()) + .addParameter("databaseId", spannerResourceManager.getDatabaseId()) + .addParameter("tableId", tableId)); + + // Act + LOG.info("Launching template with options..."); + PipelineLauncher.LaunchInfo info = launchTemplate(options); + LOG.info("Template launched. LaunchInfo: {}", info); + assertThatPipeline(info).isRunning(); + + LOG.info("Publishing messages into the Kafka topic..."); + List> expectedRecords = new ArrayList<>(); + try (KafkaProducer kafkaProducer = + kafkaResourceManager.buildProducer(new StringSerializer(), new StringSerializer())) { + for (int i = 1; i <= 10; i++) { + long id1 = Long.parseLong(i + "1"); + long id2 = Long.parseLong(i + "2"); + publish( + kafkaProducer, + topicName, + String.valueOf(id1), + "{\"id\": " + id1 + ", \"name\": \"Dataflow\"}"); + publish( + kafkaProducer, + topicName, + String.valueOf(id2), + "{\"id\": " + id2 + ", \"name\": \"Spanner\"}"); + expectedRecords.add(Map.of("id", id1, "name", "Dataflow")); + expectedRecords.add(Map.of("id", id2, "name", "Spanner")); + } + } + + LOG.info("Waiting for pipeline condition..."); + PipelineOperator.Result result = + pipelineOperator() + .waitForConditionAndFinish( + createConfig(info), + SpannerRowsCheck.builder(spannerResourceManager, tableId).setMinRows(20).build()); + + // Assert + assertThatResult(result).meetsConditions(); + + LOG.info("Verifying 20 rows in Spanner table..."); + List records = spannerResourceManager.readTableRecords(tableId, List.of("id", "name")); + assertThat(records).hasSize(20); + assertThatStructs(records).hasRecordsUnorderedCaseInsensitiveColumns(expectedRecords); + } + + private void publish( + KafkaProducer producer, String topicName, String key, String value) { + try { + RecordMetadata recordMetadata = + producer.send(new ProducerRecord<>(topicName, key, value)).get(); + LOG.info( + "Published record {}, partition {} - offset: {}", + recordMetadata.topic(), + recordMetadata.partition(), + recordMetadata.offset()); + LOG.info("Published record with key: {}, value: {}", key, value); + } catch (Exception e) { + throw new RuntimeException("Error publishing record to Kafka", e); + } + } +} From c5e0358aea50ca82bc08330eff839920625a7008 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 6 Oct 2026 19:35:48 +0000 Subject: [PATCH 2/6] regen --- yaml/README_Kafka_To_Iceberg_Yaml.md | 32 ++++--------------- yaml/README_Kafka_to_BigQuery_Yaml.md | 32 ++++--------------- yaml/README_Parquet_To_Lakehouse_Yaml.md | 2 +- .../templates/yaml/KafkaToIcebergYaml.java | 21 ++++-------- 4 files changed, 19 insertions(+), 68 deletions(-) diff --git a/yaml/README_Kafka_To_Iceberg_Yaml.md b/yaml/README_Kafka_To_Iceberg_Yaml.md index 9346638a26..46c48bc713 100644 --- a/yaml/README_Kafka_To_Iceberg_Yaml.md +++ b/yaml/README_Kafka_To_Iceberg_Yaml.md @@ -23,17 +23,13 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters -* **allowDuplicates**: If the Kafka read allows duplicates. For example: true For example, `true`. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. * **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. * **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. -* **offsetDeduplication**: If the redistribute is using offset deduplication mode. For example: true For example, `true`. -* **redistributeByRecordKey**: If the redistribute keys by the Kafka record key. For example: true For example, `true`. -* **redistributeNumKeys**: The number of keys for redistributing Kafka inputs. For example: 10 For example, `10`. -* **redistributed**: If the Kafka read should be redistributed. For example: true For example, `true`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **configProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. * **drop**: A list of field names to drop. Mutually exclusive with 'keep' and 'only'. For example, `["field_to_drop_1", "field_to_drop_2"]`. @@ -143,17 +139,13 @@ export CATALOG_PROPERTIES= export TRIGGERING_FREQUENCY_SECONDS= ### Optional -export ALLOW_DUPLICATES= export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= -export OFFSET_DEDUPLICATION= -export REDISTRIBUTE_BY_RECORD_KEY= -export REDISTRIBUTE_NUM_KEYS= -export REDISTRIBUTED= export SCHEMA= export CONFIG_PROPERTIES= export DROP= @@ -170,17 +162,13 @@ gcloud dataflow flex-template run "kafka-to-iceberg-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ - --parameters "allowDuplicates=$ALLOW_DUPLICATES" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ --parameters "messageName=$MESSAGE_NAME" \ - --parameters "offsetDeduplication=$OFFSET_DEDUPLICATION" \ - --parameters "redistributeByRecordKey=$REDISTRIBUTE_BY_RECORD_KEY" \ - --parameters "redistributeNumKeys=$REDISTRIBUTE_NUM_KEYS" \ - --parameters "redistributed=$REDISTRIBUTED" \ --parameters "schema=$SCHEMA" \ --parameters "table=$TABLE" \ --parameters "catalogName=$CATALOG_NAME" \ @@ -220,17 +208,13 @@ export CATALOG_PROPERTIES= export TRIGGERING_FREQUENCY_SECONDS= ### Optional -export ALLOW_DUPLICATES= export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= -export OFFSET_DEDUPLICATION= -export REDISTRIBUTE_BY_RECORD_KEY= -export REDISTRIBUTE_NUM_KEYS= -export REDISTRIBUTED= export SCHEMA= export CONFIG_PROPERTIES= export DROP= @@ -248,7 +232,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-iceberg-yaml-job" \ -DtemplateName="Kafka_To_Iceberg_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,allowDuplicates=$ALLOW_DUPLICATES,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,messageName=$MESSAGE_NAME,offsetDeduplication=$OFFSET_DEDUPLICATION,redistributeByRecordKey=$REDISTRIBUTE_BY_RECORD_KEY,redistributeNumKeys=$REDISTRIBUTE_NUM_KEYS,redistributed=$REDISTRIBUTED,schema=$SCHEMA,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES,triggeringFrequencySeconds=$TRIGGERING_FREQUENCY_SECONDS,sdfCheckpointAfterDuration=$SDF_CHECKPOINT_AFTER_DURATION,sdfCheckpointAfterOutputBytes=$SDF_CHECKPOINT_AFTER_OUTPUT_BYTES" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES,triggeringFrequencySeconds=$TRIGGERING_FREQUENCY_SECONDS,sdfCheckpointAfterDuration=$SDF_CHECKPOINT_AFTER_DURATION,sdfCheckpointAfterOutputBytes=$SDF_CHECKPOINT_AFTER_OUTPUT_BYTES" \ -f yaml ``` @@ -299,17 +283,13 @@ resource "google_dataflow_flex_template_job" "kafka_to_iceberg_yaml" { catalogName = "" catalogProperties = "" triggeringFrequencySeconds = "" - # allowDuplicates = "" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" + # maxReadTimeSeconds = "" # messageName = "" - # offsetDeduplication = "" - # redistributeByRecordKey = "" - # redistributeNumKeys = "" - # redistributed = "" # schema = "" # configProperties = "" # drop = "" diff --git a/yaml/README_Kafka_to_BigQuery_Yaml.md b/yaml/README_Kafka_to_BigQuery_Yaml.md index 03409b1b16..cf8c8e1928 100644 --- a/yaml/README_Kafka_to_BigQuery_Yaml.md +++ b/yaml/README_Kafka_to_BigQuery_Yaml.md @@ -28,17 +28,13 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters -* **allowDuplicates**: If the Kafka read allows duplicates. For example: true For example, `true`. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. * **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. * **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. -* **offsetDeduplication**: If the redistribute is using offset deduplication mode. For example: true For example, `true`. -* **redistributeByRecordKey**: If the redistribute keys by the Kafka record key. For example: true For example, `true`. -* **redistributeNumKeys**: The number of keys for redistributing Kafka inputs. For example: 10 For example, `10`. -* **redistributed**: If the Kafka read should be redistributed. For example: true For example, `true`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **createDisposition**: Specifies whether a table should be created if it does not exist. Valid inputs are 'CREATE_NEVER' and 'CREATE_IF_NEEDED'. Defaults to: CREATE_NEVER. * **writeDisposition**: How to specify if a write should append to an existing table, replace the table, or verify that the table is empty. Note that the my_dataset being written to must already exist. Unbounded collections can only be written using 'WRITE_EMPTY' or 'WRITE_APPEND'. Defaults to: WRITE_APPEND. @@ -141,17 +137,13 @@ export TABLE= export OUTPUT_DEADLETTER_TABLE= ### Optional -export ALLOW_DUPLICATES= export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= -export OFFSET_DEDUPLICATION= -export REDISTRIBUTE_BY_RECORD_KEY= -export REDISTRIBUTE_NUM_KEYS= -export REDISTRIBUTED= export SCHEMA= export CREATE_DISPOSITION=CREATE_NEVER export WRITE_DISPOSITION=WRITE_APPEND @@ -163,17 +155,13 @@ gcloud dataflow flex-template run "kafka-to-bigquery-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ - --parameters "allowDuplicates=$ALLOW_DUPLICATES" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ --parameters "messageName=$MESSAGE_NAME" \ - --parameters "offsetDeduplication=$OFFSET_DEDUPLICATION" \ - --parameters "redistributeByRecordKey=$REDISTRIBUTE_BY_RECORD_KEY" \ - --parameters "redistributeNumKeys=$REDISTRIBUTE_NUM_KEYS" \ - --parameters "redistributed=$REDISTRIBUTED" \ --parameters "schema=$SCHEMA" \ --parameters "table=$TABLE" \ --parameters "createDisposition=$CREATE_DISPOSITION" \ @@ -204,17 +192,13 @@ export TABLE=
export OUTPUT_DEADLETTER_TABLE= ### Optional -export ALLOW_DUPLICATES= export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= -export OFFSET_DEDUPLICATION= -export REDISTRIBUTE_BY_RECORD_KEY= -export REDISTRIBUTE_NUM_KEYS= -export REDISTRIBUTED= export SCHEMA= export CREATE_DISPOSITION=CREATE_NEVER export WRITE_DISPOSITION=WRITE_APPEND @@ -227,7 +211,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-bigquery-yaml-job" \ -DtemplateName="Kafka_to_BigQuery_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,allowDuplicates=$ALLOW_DUPLICATES,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,messageName=$MESSAGE_NAME,offsetDeduplication=$OFFSET_DEDUPLICATION,redistributeByRecordKey=$REDISTRIBUTE_BY_RECORD_KEY,redistributeNumKeys=$REDISTRIBUTE_NUM_KEYS,redistributed=$REDISTRIBUTED,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ -f yaml ``` @@ -276,17 +260,13 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigquery_yaml" { topic = "" table = "
" outputDeadletterTable = "" - # allowDuplicates = "" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" + # maxReadTimeSeconds = "" # messageName = "" - # offsetDeduplication = "" - # redistributeByRecordKey = "" - # redistributeNumKeys = "" - # redistributed = "" # schema = "" # createDisposition = "CREATE_NEVER" # writeDisposition = "WRITE_APPEND" diff --git a/yaml/README_Parquet_To_Lakehouse_Yaml.md b/yaml/README_Parquet_To_Lakehouse_Yaml.md index 50f0cb4fd3..5f7844c9ca 100644 --- a/yaml/README_Parquet_To_Lakehouse_Yaml.md +++ b/yaml/README_Parquet_To_Lakehouse_Yaml.md @@ -21,7 +21,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters -* **gcsFilePath**: An optional Google Cloud Storage directory path to copy the matched Parquet files into before registering them in the Lakehouse table. For example, `gs://your-bucket/warehouse/data/`. +* **gcsFilePath**: An optional Google Cloud Storage directory path to copy the matched Parquet files into before registering them in the Lakehouse table. This will not be used if the files are already in GCS. For example, `gs://your-bucket/warehouse/data/`. * **lakehouseCatalogProperties**: A map of properties for setting up the Lakehouse catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. * **lakehouseConfigProperties**: A map of properties to pass to the Hadoop Configuration. For example, `{"fs.gs.impl": "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem"}`. * **lakehousePartitionFields**: A list of fields and transforms for partitioning, e.g., ['day(ts)', 'category']. For example, `["day(ts)", "bucket(id, 4)"]`. diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java index 072650202c..af9374040e 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java @@ -200,15 +200,6 @@ public interface KafkaToIcebergYaml { @TemplateParameter.Text( order = 16, - name = "filter", - optional = true, - description = "An optional filter expression to apply to the input records.", - helpText = "A filter expression to apply to records from the Iceberg table.", - example = "age > 18") - String getFilter(); - - @TemplateParameter.Text( - order = 17, name = "keep", optional = true, description = "A list of field names to keep in the input record.", @@ -217,7 +208,7 @@ public interface KafkaToIcebergYaml { String getKeep(); @TemplateParameter.Text( - order = 18, + order = 17, name = "only", optional = true, description = "The name of a single record field that should be written.", @@ -226,7 +217,7 @@ public interface KafkaToIcebergYaml { String getOnly(); @TemplateParameter.Text( - order = 19, + order = 18, name = "partitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", @@ -235,7 +226,7 @@ public interface KafkaToIcebergYaml { String getPartitionFields(); @TemplateParameter.Text( - order = 20, + order = 19, name = "tableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", @@ -244,7 +235,7 @@ public interface KafkaToIcebergYaml { String getTableProperties(); @TemplateParameter.Integer( - order = 21, + order = 20, name = "triggeringFrequencySeconds", optional = false, description = "For a streaming pipeline, the frequency at which snapshots are produced.", @@ -254,7 +245,7 @@ public interface KafkaToIcebergYaml { Integer getTriggeringFrequencySeconds(); @TemplateParameter.Text( - order = 22, + order = 21, name = "sdfCheckpointAfterDuration", optional = true, description = "Dataflow Pipeline Option: Duration after which to checkpoint stateful DoFns.", @@ -265,7 +256,7 @@ public interface KafkaToIcebergYaml { String getSdfCheckpointAfterDuration(); @TemplateParameter.Integer( - order = 23, + order = 22, name = "sdfCheckpointAfterOutputBytes", optional = true, description = From c8b97a1dfbf1b16acf558f0f661d65211b0a6768 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 13 Oct 2026 15:01:24 +0000 Subject: [PATCH 3/6] add max_read_time_seconds and regen --- yaml/README_Kafka_To_Bigtable_Yaml.md | 7 ++++- yaml/README_Oracle_To_Iceberg_Yaml.md | 7 ++--- .../templates/yaml/KafkaToBigQueryYaml.java | 14 +++++----- .../templates/yaml/KafkaToBigtableYaml.java | 25 +++++++++++------ .../templates/yaml/KafkaToIcebergYaml.java | 28 +++++++++---------- yaml/src/main/yaml/KafkaToBigQuery.yaml | 1 + yaml/src/main/yaml/KafkaToBigtable.yaml | 1 + yaml/src/main/yaml/KafkaToIceberg.yaml | 1 + 8 files changed, 50 insertions(+), 34 deletions(-) diff --git a/yaml/README_Kafka_To_Bigtable_Yaml.md b/yaml/README_Kafka_To_Bigtable_Yaml.md index e00d2839fb..511d9ab86a 100644 --- a/yaml/README_Kafka_To_Bigtable_Yaml.md +++ b/yaml/README_Kafka_To_Bigtable_Yaml.md @@ -33,6 +33,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. * **fileDescriptorPath**: The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb For example, `gs://bucket/path/to/descriptor.pb`. * **format**: The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON For example, `JSON`. Defaults to: JSON. +* **maxReadTimeSeconds**: Upper bound of how long to read from Kafka in seconds. For example: 60 For example, `60`. * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **language**: The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic. For example, `python`. @@ -142,6 +143,7 @@ export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export LANGUAGE=generic @@ -157,6 +159,7 @@ gcloud dataflow flex-template run "kafka-to-bigtable-yaml-job" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ --parameters "fileDescriptorPath=$FILE_DESCRIPTOR_PATH" \ --parameters "format=$FORMAT" \ + --parameters "maxReadTimeSeconds=$MAX_READ_TIME_SECONDS" \ --parameters "messageName=$MESSAGE_NAME" \ --parameters "schema=$SCHEMA" \ --parameters "language=$LANGUAGE" \ @@ -197,6 +200,7 @@ export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= export FILE_DESCRIPTOR_PATH= export FORMAT=JSON +export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export LANGUAGE=generic @@ -208,7 +212,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-bigtable-yaml-job" \ -DtemplateName="Kafka_To_Bigtable_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,messageName=$MESSAGE_NAME,schema=$SCHEMA,language=$LANGUAGE,fields=$FIELDS,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,tableId=$TABLE_ID,windowing=$WINDOWING" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,language=$LANGUAGE,fields=$FIELDS,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,tableId=$TABLE_ID,windowing=$WINDOWING" \ -f yaml ``` @@ -265,6 +269,7 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigtable_yaml" { # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" + # maxReadTimeSeconds = "" # messageName = "" # schema = "" # language = "generic" diff --git a/yaml/README_Oracle_To_Iceberg_Yaml.md b/yaml/README_Oracle_To_Iceberg_Yaml.md index 5b22934472..3759e334d2 100644 --- a/yaml/README_Oracle_To_Iceberg_Yaml.md +++ b/yaml/README_Oracle_To_Iceberg_Yaml.md @@ -1,9 +1,8 @@ Oracle to Iceberg (YAML) template --- -The Oracle to Iceberg template is a batch pipeline executes the user provided -SQL query to read data from Oracle table and outputs the records to Iceberg -table. +The Oracle to Iceberg template is a batch pipeline executes the user provided SQL +query to read data from Oracle table and outputs the records to Iceberg table. @@ -15,7 +14,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Required parameters -* **jdbcUrl**: The JDBC connection URL. For example, `jdbc:oracle:thin:@your-host:1521:your-sid`. +* **jdbcUrl**: The JDBC connection URL. For example, `jdbc:oracle://your-host:1521/serviceName`. * **table**: A fully-qualified table identifier, e.g., my_dataset.my_table. For example, `my_dataset.my_table`. * **catalogName**: The name of the Iceberg catalog that contains the table. For example, `my_hadoop_catalog`. * **catalogProperties**: A map of properties for setting up the Iceberg catalog. For example, `{"type": "hadoop", "warehouse": "gs://your-bucket/warehouse"}`. diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java index a066bfc068..518794ebbb 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java @@ -126,7 +126,7 @@ public interface KafkaToBigQueryYaml { Integer getMaxReadTimeSeconds(); @TemplateParameter.Text( - order = 8, + order = 9, name = "messageName", optional = true, description = @@ -137,7 +137,7 @@ public interface KafkaToBigQueryYaml { String getMessageName(); @TemplateParameter.Text( - order = 9, + order = 10, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -148,7 +148,7 @@ public interface KafkaToBigQueryYaml { String getSchema(); @TemplateParameter.Text( - order = 10, + order = 11, name = "table", optional = false, description = "BigQuery table", @@ -159,7 +159,7 @@ public interface KafkaToBigQueryYaml { String getTable(); @TemplateParameter.Text( - order = 11, + order = 12, name = "createDisposition", optional = true, description = "How to create", @@ -170,7 +170,7 @@ public interface KafkaToBigQueryYaml { String getCreateDisposition(); @TemplateParameter.Text( - order = 12, + order = 13, name = "writeDisposition", optional = true, description = "How to write", @@ -181,7 +181,7 @@ public interface KafkaToBigQueryYaml { String getWriteDisposition(); @TemplateParameter.Integer( - order = 13, + order = 14, name = "numStreams", optional = true, description = "Number of streams for BigQuery Storage Write API", @@ -192,7 +192,7 @@ public interface KafkaToBigQueryYaml { Integer getNumStreams(); @TemplateParameter.Text( - order = 14, + order = 15, name = "outputDeadletterTable", optional = false, description = "The dead-letter table name to output failed messages to BigQuery", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java index 7a241663a6..efc9703b88 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java @@ -116,8 +116,17 @@ public interface KafkaToBigtableYaml { @Default.String("JSON") String getFormat(); - @TemplateParameter.Text( + @TemplateParameter.Integer( order = 8, + name = "maxReadTimeSeconds", + optional = true, + description = "Upper bound of how long to read from Kafka.", + helpText = "Upper bound of how long to read from Kafka in seconds. For example: 60", + example = "60") + Integer getMaxReadTimeSeconds(); + + @TemplateParameter.Text( + order = 9, name = "messageName", optional = true, description = @@ -128,7 +137,7 @@ public interface KafkaToBigtableYaml { String getMessageName(); @TemplateParameter.Text( - order = 9, + order = 10, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -139,7 +148,7 @@ public interface KafkaToBigtableYaml { String getSchema(); @TemplateParameter.Text( - order = 10, + order = 11, name = "language", optional = true, description = "Language used to define the expressions.", @@ -150,7 +159,7 @@ public interface KafkaToBigtableYaml { String getLanguage(); @TemplateParameter.Text( - order = 11, + order = 12, name = "fields", optional = false, description = "Field mapping configuration", @@ -161,7 +170,7 @@ public interface KafkaToBigtableYaml { String getFields(); @TemplateParameter.Text( - order = 12, + order = 13, name = "projectId", optional = false, description = "Bigtable project ID", @@ -171,7 +180,7 @@ public interface KafkaToBigtableYaml { String getProjectId(); @TemplateParameter.Text( - order = 13, + order = 14, name = "instanceId", optional = false, description = "Bigtable instance ID", @@ -181,7 +190,7 @@ public interface KafkaToBigtableYaml { String getInstanceId(); @TemplateParameter.Text( - order = 14, + order = 15, name = "tableId", optional = false, description = "Bigtable output table", @@ -191,7 +200,7 @@ public interface KafkaToBigtableYaml { String getTableId(); @TemplateParameter.Text( - order = 15, + order = 16, name = "windowing", optional = false, description = "Windowing options", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java index 6fbd7e718f..af9374040e 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java @@ -129,7 +129,7 @@ public interface KafkaToIcebergYaml { Integer getMaxReadTimeSeconds(); @TemplateParameter.Text( - order = 8, + order = 9, name = "messageName", optional = true, description = @@ -140,7 +140,7 @@ public interface KafkaToIcebergYaml { String getMessageName(); @TemplateParameter.Text( - order = 9, + order = 10, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -151,7 +151,7 @@ public interface KafkaToIcebergYaml { String getSchema(); @TemplateParameter.Text( - order = 10, + order = 11, name = "table", optional = false, description = "A fully-qualified table identifier.", @@ -161,7 +161,7 @@ public interface KafkaToIcebergYaml { String getTable(); @TemplateParameter.Text( - order = 11, + order = 12, name = "catalogName", optional = false, description = "Name of the catalog containing the table.", @@ -171,7 +171,7 @@ public interface KafkaToIcebergYaml { String getCatalogName(); @TemplateParameter.Text( - order = 12, + order = 13, name = "catalogProperties", optional = false, description = "Properties used to set up the Iceberg catalog.", @@ -181,7 +181,7 @@ public interface KafkaToIcebergYaml { String getCatalogProperties(); @TemplateParameter.Text( - order = 13, + order = 14, name = "configProperties", optional = true, description = "Properties passed to the Hadoop Configuration.", @@ -190,7 +190,7 @@ public interface KafkaToIcebergYaml { String getConfigProperties(); @TemplateParameter.Text( - order = 14, + order = 15, name = "drop", optional = true, description = "A list of field names to drop from the input record before writing.", @@ -199,7 +199,7 @@ public interface KafkaToIcebergYaml { String getDrop(); @TemplateParameter.Text( - order = 15, + order = 16, name = "keep", optional = true, description = "A list of field names to keep in the input record.", @@ -208,7 +208,7 @@ public interface KafkaToIcebergYaml { String getKeep(); @TemplateParameter.Text( - order = 16, + order = 17, name = "only", optional = true, description = "The name of a single record field that should be written.", @@ -217,7 +217,7 @@ public interface KafkaToIcebergYaml { String getOnly(); @TemplateParameter.Text( - order = 17, + order = 18, name = "partitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", @@ -226,7 +226,7 @@ public interface KafkaToIcebergYaml { String getPartitionFields(); @TemplateParameter.Text( - order = 18, + order = 19, name = "tableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", @@ -235,7 +235,7 @@ public interface KafkaToIcebergYaml { String getTableProperties(); @TemplateParameter.Integer( - order = 19, + order = 20, name = "triggeringFrequencySeconds", optional = false, description = "For a streaming pipeline, the frequency at which snapshots are produced.", @@ -245,7 +245,7 @@ public interface KafkaToIcebergYaml { Integer getTriggeringFrequencySeconds(); @TemplateParameter.Text( - order = 20, + order = 21, name = "sdfCheckpointAfterDuration", optional = true, description = "Dataflow Pipeline Option: Duration after which to checkpoint stateful DoFns.", @@ -256,7 +256,7 @@ public interface KafkaToIcebergYaml { String getSdfCheckpointAfterDuration(); @TemplateParameter.Integer( - order = 21, + order = 22, name = "sdfCheckpointAfterOutputBytes", optional = true, description = diff --git a/yaml/src/main/yaml/KafkaToBigQuery.yaml b/yaml/src/main/yaml/KafkaToBigQuery.yaml index d06fa6ad5a..2d44063f83 100644 --- a/yaml/src/main/yaml/KafkaToBigQuery.yaml +++ b/yaml/src/main/yaml/KafkaToBigQuery.yaml @@ -62,6 +62,7 @@ pipeline: format: {{ format | default('JSON') }} topic: {{ topic }} bootstrap_servers: {{ bootstrapServers }} + max_read_time_seconds: {{ maxReadTimeSeconds }} auto_offset_reset_config: 'earliest' error_handling: output: errors diff --git a/yaml/src/main/yaml/KafkaToBigtable.yaml b/yaml/src/main/yaml/KafkaToBigtable.yaml index 0dceaa22f7..a3fbcd0611 100644 --- a/yaml/src/main/yaml/KafkaToBigtable.yaml +++ b/yaml/src/main/yaml/KafkaToBigtable.yaml @@ -49,6 +49,7 @@ pipeline: consumer_config: {{ consumerConfigUpdates }} file_descriptor_path: {{ fileDescriptorPath }} format: {{ format | default('JSON') }} + max_read_time_seconds: {{ maxReadTimeSeconds }} message_name: {{ messageName }} {% if schema %} schema: | diff --git a/yaml/src/main/yaml/KafkaToIceberg.yaml b/yaml/src/main/yaml/KafkaToIceberg.yaml index 749a4ceee3..daf2d24d3f 100644 --- a/yaml/src/main/yaml/KafkaToIceberg.yaml +++ b/yaml/src/main/yaml/KafkaToIceberg.yaml @@ -57,6 +57,7 @@ pipeline: consumer_config: {{ consumerConfigUpdates }} file_descriptor_path: {{ fileDescriptorPath }} format: {{ format | default('JSON') }} + max_read_time_seconds: {{ maxReadTimeSeconds }} message_name: {{ messageName }} {% if schema %} schema: | From 09db6e961c423ae9deeaa8d73b59d7e5d6a928c9 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 13 Oct 2026 15:15:43 +0000 Subject: [PATCH 4/6] update auto_offset_reset_config to be a jinja parameter --- yaml/README_Kafka_To_Bigtable_Yaml.md | 7 ++- yaml/README_Kafka_To_Iceberg_Yaml.md | 7 ++- yaml/README_Kafka_To_Spanner_Yaml.md | 7 ++- yaml/README_Kafka_to_BigQuery_Yaml.md | 7 ++- .../templates/yaml/KafkaToBigQueryYaml.java | 36 ++++++++----- .../templates/yaml/KafkaToBigtableYaml.java | 38 +++++++++----- .../templates/yaml/KafkaToIcebergYaml.java | 50 ++++++++++++------- .../templates/yaml/KafkaToSpannerYaml.java | 34 +++++++++---- .../main/python/options/kafka_options.yaml | 22 +++++--- yaml/src/main/yaml/KafkaToBigQuery.yaml | 2 +- yaml/src/main/yaml/KafkaToBigtable.yaml | 2 +- yaml/src/main/yaml/KafkaToIceberg.yaml | 2 +- yaml/src/main/yaml/KafkaToSpanner.yaml | 2 +- 13 files changed, 146 insertions(+), 70 deletions(-) diff --git a/yaml/README_Kafka_To_Bigtable_Yaml.md b/yaml/README_Kafka_To_Bigtable_Yaml.md index 511d9ab86a..2776cc2428 100644 --- a/yaml/README_Kafka_To_Bigtable_Yaml.md +++ b/yaml/README_Kafka_To_Bigtable_Yaml.md @@ -28,6 +28,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. @@ -138,6 +139,7 @@ export TABLE_ID= export WINDOWING= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -154,6 +156,7 @@ gcloud dataflow flex-template run "kafka-to-bigtable-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ @@ -195,6 +198,7 @@ export TABLE_ID= export WINDOWING= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -212,7 +216,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-bigtable-yaml-job" \ -DtemplateName="Kafka_To_Bigtable_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,language=$LANGUAGE,fields=$FIELDS,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,tableId=$TABLE_ID,windowing=$WINDOWING" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,language=$LANGUAGE,fields=$FIELDS,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,tableId=$TABLE_ID,windowing=$WINDOWING" \ -f yaml ``` @@ -264,6 +268,7 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigtable_yaml" { instanceId = "" tableId = "" windowing = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" diff --git a/yaml/README_Kafka_To_Iceberg_Yaml.md b/yaml/README_Kafka_To_Iceberg_Yaml.md index 46c48bc713..9cf6ec244e 100644 --- a/yaml/README_Kafka_To_Iceberg_Yaml.md +++ b/yaml/README_Kafka_To_Iceberg_Yaml.md @@ -23,6 +23,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. @@ -139,6 +140,7 @@ export CATALOG_PROPERTIES= export TRIGGERING_FREQUENCY_SECONDS= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -162,6 +164,7 @@ gcloud dataflow flex-template run "kafka-to-iceberg-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ @@ -208,6 +211,7 @@ export CATALOG_PROPERTIES= export TRIGGERING_FREQUENCY_SECONDS= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -232,7 +236,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-iceberg-yaml-job" \ -DtemplateName="Kafka_To_Iceberg_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES,triggeringFrequencySeconds=$TRIGGERING_FREQUENCY_SECONDS,sdfCheckpointAfterDuration=$SDF_CHECKPOINT_AFTER_DURATION,sdfCheckpointAfterOutputBytes=$SDF_CHECKPOINT_AFTER_OUTPUT_BYTES" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,catalogName=$CATALOG_NAME,catalogProperties=$CATALOG_PROPERTIES,configProperties=$CONFIG_PROPERTIES,drop=$DROP,keep=$KEEP,only=$ONLY,partitionFields=$PARTITION_FIELDS,tableProperties=$TABLE_PROPERTIES,triggeringFrequencySeconds=$TRIGGERING_FREQUENCY_SECONDS,sdfCheckpointAfterDuration=$SDF_CHECKPOINT_AFTER_DURATION,sdfCheckpointAfterOutputBytes=$SDF_CHECKPOINT_AFTER_OUTPUT_BYTES" \ -f yaml ``` @@ -283,6 +287,7 @@ resource "google_dataflow_flex_template_job" "kafka_to_iceberg_yaml" { catalogName = "" catalogProperties = "" triggeringFrequencySeconds = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" diff --git a/yaml/README_Kafka_To_Spanner_Yaml.md b/yaml/README_Kafka_To_Spanner_Yaml.md index 1517bee210..7e8b8c6513 100644 --- a/yaml/README_Kafka_To_Spanner_Yaml.md +++ b/yaml/README_Kafka_To_Spanner_Yaml.md @@ -25,6 +25,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. @@ -133,6 +134,7 @@ export DATABASE_ID= export TABLE_ID= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -149,6 +151,7 @@ gcloud dataflow flex-template run "kafka-to-spanner-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ @@ -186,6 +189,7 @@ export DATABASE_ID= export TABLE_ID= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -203,7 +207,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-spanner-yaml-job" \ -DtemplateName="Kafka_To_Spanner_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,databaseId=$DATABASE_ID,tableId=$TABLE_ID" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,databaseId=$DATABASE_ID,tableId=$TABLE_ID" \ -f yaml ``` @@ -253,6 +257,7 @@ resource "google_dataflow_flex_template_job" "kafka_to_spanner_yaml" { instanceId = "" databaseId = "" tableId = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" diff --git a/yaml/README_Kafka_to_BigQuery_Yaml.md b/yaml/README_Kafka_to_BigQuery_Yaml.md index 60514e205a..a39d98de15 100644 --- a/yaml/README_Kafka_to_BigQuery_Yaml.md +++ b/yaml/README_Kafka_to_BigQuery_Yaml.md @@ -28,6 +28,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat ### Optional parameters +* **autoOffsetResetConfig**: What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest For example, `earliest`. Defaults to: earliest. * **confluentSchemaRegistrySubject**: The subject name for the Confluent Schema Registry. For example: my_subject For example, `my_subject`. * **confluentSchemaRegistryUrl**: The URL for the Confluent Schema Registry. For example: http://schema-registry:8081 For example, `http://schema-registry:8081`. * **consumerConfigUpdates**: A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'} For example, `{"group.id": "my_group"}`. @@ -137,6 +138,7 @@ export TABLE=
export OUTPUT_DEADLETTER_TABLE= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -155,6 +157,7 @@ gcloud dataflow flex-template run "kafka-to-bigquery-yaml-job" \ --template-file-gcs-location "$TEMPLATE_SPEC_GCSPATH" \ --parameters "bootstrapServers=$BOOTSTRAP_SERVERS" \ --parameters "topic=$TOPIC" \ + --parameters "autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG" \ --parameters "confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT" \ --parameters "confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL" \ --parameters "consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES" \ @@ -192,6 +195,7 @@ export TABLE=
export OUTPUT_DEADLETTER_TABLE= ### Optional +export AUTO_OFFSET_RESET_CONFIG=earliest export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= export CONFLUENT_SCHEMA_REGISTRY_URL= export CONSUMER_CONFIG_UPDATES= @@ -211,7 +215,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-bigquery-yaml-job" \ -DtemplateName="Kafka_to_BigQuery_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ -f yaml ``` @@ -260,6 +264,7 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigquery_yaml" { topic = "" table = "
" outputDeadletterTable = "" + # autoOffsetResetConfig = "earliest" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java index 518794ebbb..29d3e07e54 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigQueryYaml.java @@ -67,6 +67,18 @@ public interface KafkaToBigQueryYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -75,7 +87,7 @@ public interface KafkaToBigQueryYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -85,7 +97,7 @@ public interface KafkaToBigQueryYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -96,7 +108,7 @@ public interface KafkaToBigQueryYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -106,7 +118,7 @@ public interface KafkaToBigQueryYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -117,7 +129,7 @@ public interface KafkaToBigQueryYaml { String getFormat(); @TemplateParameter.Integer( - order = 8, + order = 9, name = "maxReadTimeSeconds", optional = true, description = "Upper bound of how long to read from Kafka.", @@ -126,7 +138,7 @@ public interface KafkaToBigQueryYaml { Integer getMaxReadTimeSeconds(); @TemplateParameter.Text( - order = 9, + order = 10, name = "messageName", optional = true, description = @@ -137,7 +149,7 @@ public interface KafkaToBigQueryYaml { String getMessageName(); @TemplateParameter.Text( - order = 10, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -148,7 +160,7 @@ public interface KafkaToBigQueryYaml { String getSchema(); @TemplateParameter.Text( - order = 11, + order = 12, name = "table", optional = false, description = "BigQuery table", @@ -159,7 +171,7 @@ public interface KafkaToBigQueryYaml { String getTable(); @TemplateParameter.Text( - order = 12, + order = 13, name = "createDisposition", optional = true, description = "How to create", @@ -170,7 +182,7 @@ public interface KafkaToBigQueryYaml { String getCreateDisposition(); @TemplateParameter.Text( - order = 13, + order = 14, name = "writeDisposition", optional = true, description = "How to write", @@ -181,7 +193,7 @@ public interface KafkaToBigQueryYaml { String getWriteDisposition(); @TemplateParameter.Integer( - order = 14, + order = 15, name = "numStreams", optional = true, description = "Number of streams for BigQuery Storage Write API", @@ -192,7 +204,7 @@ public interface KafkaToBigQueryYaml { Integer getNumStreams(); @TemplateParameter.Text( - order = 15, + order = 16, name = "outputDeadletterTable", optional = false, description = "The dead-letter table name to output failed messages to BigQuery", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java index efc9703b88..9f06355b52 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java @@ -67,6 +67,18 @@ public interface KafkaToBigtableYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -75,7 +87,7 @@ public interface KafkaToBigtableYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -85,7 +97,7 @@ public interface KafkaToBigtableYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -96,7 +108,7 @@ public interface KafkaToBigtableYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -106,7 +118,7 @@ public interface KafkaToBigtableYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -117,7 +129,7 @@ public interface KafkaToBigtableYaml { String getFormat(); @TemplateParameter.Integer( - order = 8, + order = 9, name = "maxReadTimeSeconds", optional = true, description = "Upper bound of how long to read from Kafka.", @@ -126,7 +138,7 @@ public interface KafkaToBigtableYaml { Integer getMaxReadTimeSeconds(); @TemplateParameter.Text( - order = 9, + order = 10, name = "messageName", optional = true, description = @@ -137,7 +149,7 @@ public interface KafkaToBigtableYaml { String getMessageName(); @TemplateParameter.Text( - order = 10, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -148,7 +160,7 @@ public interface KafkaToBigtableYaml { String getSchema(); @TemplateParameter.Text( - order = 11, + order = 12, name = "language", optional = true, description = "Language used to define the expressions.", @@ -159,7 +171,7 @@ public interface KafkaToBigtableYaml { String getLanguage(); @TemplateParameter.Text( - order = 12, + order = 13, name = "fields", optional = false, description = "Field mapping configuration", @@ -170,7 +182,7 @@ public interface KafkaToBigtableYaml { String getFields(); @TemplateParameter.Text( - order = 13, + order = 14, name = "projectId", optional = false, description = "Bigtable project ID", @@ -180,7 +192,7 @@ public interface KafkaToBigtableYaml { String getProjectId(); @TemplateParameter.Text( - order = 14, + order = 15, name = "instanceId", optional = false, description = "Bigtable instance ID", @@ -190,7 +202,7 @@ public interface KafkaToBigtableYaml { String getInstanceId(); @TemplateParameter.Text( - order = 15, + order = 16, name = "tableId", optional = false, description = "Bigtable output table", @@ -200,7 +212,7 @@ public interface KafkaToBigtableYaml { String getTableId(); @TemplateParameter.Text( - order = 16, + order = 17, name = "windowing", optional = false, description = "Windowing options", diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java index af9374040e..07a9d9c89c 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToIcebergYaml.java @@ -70,6 +70,18 @@ public interface KafkaToIcebergYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -78,7 +90,7 @@ public interface KafkaToIcebergYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -88,7 +100,7 @@ public interface KafkaToIcebergYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -99,7 +111,7 @@ public interface KafkaToIcebergYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -109,7 +121,7 @@ public interface KafkaToIcebergYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -120,7 +132,7 @@ public interface KafkaToIcebergYaml { String getFormat(); @TemplateParameter.Integer( - order = 8, + order = 9, name = "maxReadTimeSeconds", optional = true, description = "Upper bound of how long to read from Kafka.", @@ -129,7 +141,7 @@ public interface KafkaToIcebergYaml { Integer getMaxReadTimeSeconds(); @TemplateParameter.Text( - order = 9, + order = 10, name = "messageName", optional = true, description = @@ -140,7 +152,7 @@ public interface KafkaToIcebergYaml { String getMessageName(); @TemplateParameter.Text( - order = 10, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -151,7 +163,7 @@ public interface KafkaToIcebergYaml { String getSchema(); @TemplateParameter.Text( - order = 11, + order = 12, name = "table", optional = false, description = "A fully-qualified table identifier.", @@ -161,7 +173,7 @@ public interface KafkaToIcebergYaml { String getTable(); @TemplateParameter.Text( - order = 12, + order = 13, name = "catalogName", optional = false, description = "Name of the catalog containing the table.", @@ -171,7 +183,7 @@ public interface KafkaToIcebergYaml { String getCatalogName(); @TemplateParameter.Text( - order = 13, + order = 14, name = "catalogProperties", optional = false, description = "Properties used to set up the Iceberg catalog.", @@ -181,7 +193,7 @@ public interface KafkaToIcebergYaml { String getCatalogProperties(); @TemplateParameter.Text( - order = 14, + order = 15, name = "configProperties", optional = true, description = "Properties passed to the Hadoop Configuration.", @@ -190,7 +202,7 @@ public interface KafkaToIcebergYaml { String getConfigProperties(); @TemplateParameter.Text( - order = 15, + order = 16, name = "drop", optional = true, description = "A list of field names to drop from the input record before writing.", @@ -199,7 +211,7 @@ public interface KafkaToIcebergYaml { String getDrop(); @TemplateParameter.Text( - order = 16, + order = 17, name = "keep", optional = true, description = "A list of field names to keep in the input record.", @@ -208,7 +220,7 @@ public interface KafkaToIcebergYaml { String getKeep(); @TemplateParameter.Text( - order = 17, + order = 18, name = "only", optional = true, description = "The name of a single record field that should be written.", @@ -217,7 +229,7 @@ public interface KafkaToIcebergYaml { String getOnly(); @TemplateParameter.Text( - order = 18, + order = 19, name = "partitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", @@ -226,7 +238,7 @@ public interface KafkaToIcebergYaml { String getPartitionFields(); @TemplateParameter.Text( - order = 19, + order = 20, name = "tableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", @@ -235,7 +247,7 @@ public interface KafkaToIcebergYaml { String getTableProperties(); @TemplateParameter.Integer( - order = 20, + order = 21, name = "triggeringFrequencySeconds", optional = false, description = "For a streaming pipeline, the frequency at which snapshots are produced.", @@ -245,7 +257,7 @@ public interface KafkaToIcebergYaml { Integer getTriggeringFrequencySeconds(); @TemplateParameter.Text( - order = 21, + order = 22, name = "sdfCheckpointAfterDuration", optional = true, description = "Dataflow Pipeline Option: Duration after which to checkpoint stateful DoFns.", @@ -256,7 +268,7 @@ public interface KafkaToIcebergYaml { String getSdfCheckpointAfterDuration(); @TemplateParameter.Integer( - order = 22, + order = 23, name = "sdfCheckpointAfterOutputBytes", optional = true, description = diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java index 3ff1404866..894dbb15e8 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java @@ -67,6 +67,18 @@ public interface KafkaToSpannerYaml { @TemplateParameter.Text( order = 3, + name = "autoOffsetResetConfig", + optional = true, + description = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server.", + helpText = + "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest", + example = "earliest") + @Default.String("earliest") + String getAutoOffsetResetConfig(); + + @TemplateParameter.Text( + order = 4, name = "confluentSchemaRegistrySubject", optional = true, description = "The subject name for the Confluent Schema Registry.", @@ -75,7 +87,7 @@ public interface KafkaToSpannerYaml { String getConfluentSchemaRegistrySubject(); @TemplateParameter.Text( - order = 4, + order = 5, name = "confluentSchemaRegistryUrl", optional = true, description = "The URL for the Confluent Schema Registry.", @@ -85,7 +97,7 @@ public interface KafkaToSpannerYaml { String getConfluentSchemaRegistryUrl(); @TemplateParameter.Text( - order = 5, + order = 6, name = "consumerConfigUpdates", optional = true, description = @@ -96,7 +108,7 @@ public interface KafkaToSpannerYaml { String getConsumerConfigUpdates(); @TemplateParameter.Text( - order = 6, + order = 7, name = "fileDescriptorPath", optional = true, description = "The path to the Protocol Buffer File Descriptor Set file.", @@ -106,7 +118,7 @@ public interface KafkaToSpannerYaml { String getFileDescriptorPath(); @TemplateParameter.Text( - order = 7, + order = 8, name = "format", optional = true, description = "The encoding format for the data stored in Kafka.", @@ -117,7 +129,7 @@ public interface KafkaToSpannerYaml { String getFormat(); @TemplateParameter.Integer( - order = 8, + order = 9, name = "maxReadTimeSeconds", optional = true, description = "Upper bound of how long to read from Kafka.", @@ -126,7 +138,7 @@ public interface KafkaToSpannerYaml { Integer getMaxReadTimeSeconds(); @TemplateParameter.Text( - order = 9, + order = 10, name = "messageName", optional = true, description = @@ -137,7 +149,7 @@ public interface KafkaToSpannerYaml { String getMessageName(); @TemplateParameter.Text( - order = 10, + order = 11, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -148,7 +160,7 @@ public interface KafkaToSpannerYaml { String getSchema(); @TemplateParameter.Text( - order = 11, + order = 12, name = "projectId", optional = true, description = "Cloud Spanner Project ID.", @@ -157,7 +169,7 @@ public interface KafkaToSpannerYaml { String getProjectId(); @TemplateParameter.Text( - order = 12, + order = 13, name = "instanceId", optional = false, description = "Cloud Spanner Instance ID.", @@ -167,7 +179,7 @@ public interface KafkaToSpannerYaml { String getInstanceId(); @TemplateParameter.Text( - order = 13, + order = 14, name = "databaseId", optional = false, description = "Cloud Spanner Database ID.", @@ -177,7 +189,7 @@ public interface KafkaToSpannerYaml { String getDatabaseId(); @TemplateParameter.Text( - order = 14, + order = 15, name = "tableId", optional = false, description = "Cloud Spanner Table ID.", diff --git a/yaml/src/main/python/options/kafka_options.yaml b/yaml/src/main/python/options/kafka_options.yaml index 2f21c53f87..fdead46144 100644 --- a/yaml/src/main/python/options/kafka_options.yaml +++ b/yaml/src/main/python/options/kafka_options.yaml @@ -16,34 +16,42 @@ options: required: true type: text - order: 3 + name: "autoOffsetResetConfig" + description: "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server." + help: "What to do when there is no initial offset in Kafka or if the current offset does not exist any more on the server. (1) earliest: automatically reset the offset to the earliest offset. (2) latest: automatically reset the offset to the latest offset (3) none: throw exception to the consumer if no previous offset is found for the consumer's group. For example: earliest" + example: "earliest" + required: false + default: "earliest" + type: text + - order: 4 name: "confluentSchemaRegistrySubject" description: "The subject name for the Confluent Schema Registry." help: "The subject name for the Confluent Schema Registry. For example: my_subject" example: "my_subject" required: false type: text - - order: 4 + - order: 5 name: "confluentSchemaRegistryUrl" description: "The URL for the Confluent Schema Registry." help: "The URL for the Confluent Schema Registry. For example: http://schema-registry:8081" example: "http://schema-registry:8081" required: false type: text - - order: 5 + - order: 6 name: "consumerConfigUpdates" description: "A list of key-value pairs that act as configuration parameters for Kafka consumers." help: "A list of key-value pairs that act as configuration parameters for Kafka consumers. For example: {'group.id': 'my_group'}" example: '{"group.id": "my_group"}' required: false type: map - - order: 6 + - order: 7 name: "fileDescriptorPath" description: "The path to the Protocol Buffer File Descriptor Set file." help: "The path to the Protocol Buffer File Descriptor Set file. For example: gs://bucket/path/to/descriptor.pb" example: "gs://bucket/path/to/descriptor.pb" required: false type: text - - order: 7 + - order: 8 name: "format" description: "The encoding format for the data stored in Kafka." help: "The encoding format for the data stored in Kafka. Valid options are: RAW,STRING,AVRO,JSON,PROTO. For example: JSON" @@ -51,21 +59,21 @@ options: required: false default: "JSON" type: text - - order: 8 + - order: 9 name: "maxReadTimeSeconds" description: "Upper bound of how long to read from Kafka." help: "Upper bound of how long to read from Kafka in seconds. For example: 60" example: "60" required: false type: integer - - order: 9 + - order: 10 name: "messageName" description: "The name of the Protocol Buffer message to be used for schema extraction and data conversion." help: "The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage" example: "MyMessage" required: false type: text - - order: 10 + - order: 11 name: "schema" description: "The schema in which the data is encoded in the Kafka topic." help: > diff --git a/yaml/src/main/yaml/KafkaToBigQuery.yaml b/yaml/src/main/yaml/KafkaToBigQuery.yaml index 2d44063f83..ecd22e5148 100644 --- a/yaml/src/main/yaml/KafkaToBigQuery.yaml +++ b/yaml/src/main/yaml/KafkaToBigQuery.yaml @@ -63,7 +63,7 @@ pipeline: topic: {{ topic }} bootstrap_servers: {{ bootstrapServers }} max_read_time_seconds: {{ maxReadTimeSeconds }} - auto_offset_reset_config: 'earliest' + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} error_handling: output: errors - type: WriteToBigQuery diff --git a/yaml/src/main/yaml/KafkaToBigtable.yaml b/yaml/src/main/yaml/KafkaToBigtable.yaml index a3fbcd0611..d7884223b1 100644 --- a/yaml/src/main/yaml/KafkaToBigtable.yaml +++ b/yaml/src/main/yaml/KafkaToBigtable.yaml @@ -55,7 +55,7 @@ pipeline: schema: | {{ schema }} {% endif %} - auto_offset_reset_config: 'earliest' + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} - type: MapToFields name: ConvertStringsToBytes input: ReadFromKafka diff --git a/yaml/src/main/yaml/KafkaToIceberg.yaml b/yaml/src/main/yaml/KafkaToIceberg.yaml index daf2d24d3f..f511681740 100644 --- a/yaml/src/main/yaml/KafkaToIceberg.yaml +++ b/yaml/src/main/yaml/KafkaToIceberg.yaml @@ -63,7 +63,7 @@ pipeline: schema: | {{ schema }} {% endif %} - auto_offset_reset_config: 'earliest' + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} - type: WriteToIceberg name: WriteToIceberg diff --git a/yaml/src/main/yaml/KafkaToSpanner.yaml b/yaml/src/main/yaml/KafkaToSpanner.yaml index e300303bfd..f85e3905d5 100644 --- a/yaml/src/main/yaml/KafkaToSpanner.yaml +++ b/yaml/src/main/yaml/KafkaToSpanner.yaml @@ -50,7 +50,7 @@ pipeline: schema: | {{ schema }} {% endif %} - auto_offset_reset_config: 'earliest' + auto_offset_reset_config: {{ autoOffsetResetConfig | default('earliest') }} - type: WriteToSpanner name: WriteToSpanner config: From 24db6f456f33b26ced285033bc991c3f4e1e76b5 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 13 Oct 2026 16:21:44 +0000 Subject: [PATCH 5/6] update windowing with jinja parameters etc --- yaml/README_Kafka_To_Spanner_Yaml.md | 16 ++++++++++--- .../templates/yaml/KafkaToSpannerYaml.java | 23 ++++++++++++++++++- .../main/python/options/spanner_options.yaml | 2 +- yaml/src/main/yaml/KafkaToSpanner.yaml | 6 +++-- 4 files changed, 40 insertions(+), 7 deletions(-) diff --git a/yaml/README_Kafka_To_Spanner_Yaml.md b/yaml/README_Kafka_To_Spanner_Yaml.md index 7e8b8c6513..508d62c4bc 100644 --- a/yaml/README_Kafka_To_Spanner_Yaml.md +++ b/yaml/README_Kafka_To_Spanner_Yaml.md @@ -21,7 +21,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat * **topic**: Kafka topic to read from. For example: my_topic For example, `my_topic`. * **instanceId**: The Cloud Spanner instance ID. For example, `your-instance-id`. * **databaseId**: The Cloud Spanner database ID. For example, `your-database-id`. -* **tableId**: The Cloud Spanner table ID to write the output to. For example, `your-table-id`. +* **tableId**: The Cloud Spanner table ID. For example, `your-table-id`. ### Optional parameters @@ -35,6 +35,8 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat * **messageName**: The name of the Protocol Buffer message to be used for schema extraction and data conversion. For example: MyMessage For example, `MyMessage`. * **schema**: The schema in which the data is encoded in the Kafka topic. For example: {'type': 'record', 'name': 'User', 'fields': [{'name': 'name', 'type': 'string'}]}. A schema is required if data format is JSON, AVRO or PROTO. For example, `{"type": "record", "name": "User", "fields": [{"name": "name", "type": "string"}]}`. * **projectId**: The Google Cloud project ID of the Cloud Spanner instance. For example, `your-project-id`. +* **type**: The type of windowing to apply (e.g., fixed, sliding, sessions, global). Defaults to fixed. For example, `fixed`. +* **size**: The window size in seconds. Defaults to 60. For example, `60`. @@ -144,6 +146,8 @@ export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export PROJECT_ID= +export TYPE=fixed +export SIZE=60 gcloud dataflow flex-template run "kafka-to-spanner-yaml-job" \ --project "$PROJECT" \ @@ -163,7 +167,9 @@ gcloud dataflow flex-template run "kafka-to-spanner-yaml-job" \ --parameters "projectId=$PROJECT_ID" \ --parameters "instanceId=$INSTANCE_ID" \ --parameters "databaseId=$DATABASE_ID" \ - --parameters "tableId=$TABLE_ID" + --parameters "tableId=$TABLE_ID" \ + --parameters "type=$TYPE" \ + --parameters "size=$SIZE" ``` For more information about the command, please check: @@ -199,6 +205,8 @@ export MAX_READ_TIME_SECONDS= export MESSAGE_NAME= export SCHEMA= export PROJECT_ID= +export TYPE=fixed +export SIZE=60 mvn clean package -PtemplatesRun \ -DskipTests \ @@ -207,7 +215,7 @@ mvn clean package -PtemplatesRun \ -Dregion="$REGION" \ -DjobName="kafka-to-spanner-yaml-job" \ -DtemplateName="Kafka_To_Spanner_Yaml" \ --Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,databaseId=$DATABASE_ID,tableId=$TABLE_ID" \ +-Dparameters="bootstrapServers=$BOOTSTRAP_SERVERS,topic=$TOPIC,autoOffsetResetConfig=$AUTO_OFFSET_RESET_CONFIG,confluentSchemaRegistrySubject=$CONFLUENT_SCHEMA_REGISTRY_SUBJECT,confluentSchemaRegistryUrl=$CONFLUENT_SCHEMA_REGISTRY_URL,consumerConfigUpdates=$CONSUMER_CONFIG_UPDATES,fileDescriptorPath=$FILE_DESCRIPTOR_PATH,format=$FORMAT,maxReadTimeSeconds=$MAX_READ_TIME_SECONDS,messageName=$MESSAGE_NAME,schema=$SCHEMA,projectId=$PROJECT_ID,instanceId=$INSTANCE_ID,databaseId=$DATABASE_ID,tableId=$TABLE_ID,type=$TYPE,size=$SIZE" \ -f yaml ``` @@ -267,6 +275,8 @@ resource "google_dataflow_flex_template_job" "kafka_to_spanner_yaml" { # messageName = "" # schema = "" # projectId = "" + # type = "fixed" + # size = "60" } } ``` diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java index 894dbb15e8..e71c517b1d 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToSpannerYaml.java @@ -193,8 +193,29 @@ public interface KafkaToSpannerYaml { name = "tableId", optional = false, description = "Cloud Spanner Table ID.", - helpText = "The Cloud Spanner table ID to write the output to.", + helpText = "The Cloud Spanner table ID.", example = "your-table-id") @Validation.Required String getTableId(); + + @TemplateParameter.Text( + order = 16, + name = "type", + optional = true, + description = "The type of windowing to apply.", + helpText = + "The type of windowing to apply (e.g., fixed, sliding, sessions, global). Defaults to fixed.", + example = "fixed") + @Default.String("fixed") + String getType(); + + @TemplateParameter.Integer( + order = 17, + name = "size", + optional = true, + description = "The window size in seconds.", + helpText = "The window size in seconds. Defaults to 60.", + example = "60") + @Default.Integer(60) + Integer getSize(); } diff --git a/yaml/src/main/python/options/spanner_options.yaml b/yaml/src/main/python/options/spanner_options.yaml index 916ab3f633..9f46860f9e 100644 --- a/yaml/src/main/python/options/spanner_options.yaml +++ b/yaml/src/main/python/options/spanner_options.yaml @@ -25,7 +25,7 @@ options: - order: 4 name: "tableId" description: "Cloud Spanner Table ID." - help: "The Cloud Spanner table ID to write the output to." + help: "The Cloud Spanner table ID." example: "your-table-id" required: true type: text diff --git a/yaml/src/main/yaml/KafkaToSpanner.yaml b/yaml/src/main/yaml/KafkaToSpanner.yaml index f85e3905d5..a640124503 100644 --- a/yaml/src/main/yaml/KafkaToSpanner.yaml +++ b/yaml/src/main/yaml/KafkaToSpanner.yaml @@ -26,10 +26,12 @@ template: options_file: - "kafka_options" - "spanner_options" + - "windowing_options" parameters: - kafka_read_options - spanner_common_options + - windowing_common_options pipeline: type: chain @@ -59,8 +61,8 @@ pipeline: database_id: {{ databaseId }} table: {{ tableId }} windowing: - type: fixed - size: 60 + type: {{ type | default('fixed') }} + size: {{ size | default(60) }} options: streaming: true From a39351969657cd9a1610afaa8d612810988a6e96 Mon Sep 17 00:00:00 2001 From: Derrick Williams Date: Tue, 13 Oct 2026 16:25:47 +0000 Subject: [PATCH 6/6] add windowing_optinos.yaml file --- .../python/options/windowing_options.yaml | 19 +++++++++++++++++++ 1 file changed, 19 insertions(+) create mode 100644 yaml/src/main/python/options/windowing_options.yaml diff --git a/yaml/src/main/python/options/windowing_options.yaml b/yaml/src/main/python/options/windowing_options.yaml new file mode 100644 index 0000000000..b695914ede --- /dev/null +++ b/yaml/src/main/python/options/windowing_options.yaml @@ -0,0 +1,19 @@ +options: + - name: "windowing_common_options" + parameters: + - order: 1 + name: "type" + description: "The type of windowing to apply." + help: "The type of windowing to apply (e.g., fixed, sliding, sessions, global). Defaults to fixed." + example: "fixed" + required: false + default: "fixed" + type: text + - order: 2 + name: "size" + description: "The window size in seconds." + help: "The window size in seconds. Defaults to 60." + example: "60" + required: false + default: 60 + type: integer