add kafka to bigtable blueprint

yaml/README_Kafka_To_Bigtable_Yaml.md

@@ -0,0 +1,273 @@
+
+Kafka to Bigtable (YAML) template
+---
+The Kafka to Bigtable template is a streaming pipeline which ingests data from an
+Apache Kafka topic, executes a user-defined mapping, and writes the resulting
+records to Bigtable.
+
+
+:memo: This is a Google-provided template! Please
+check [Provided templates documentation](https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-bigtable)
+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_Bigtable_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`.
+* **fields**: The output fields to compute, each mapping to the expression or callable that creates them. For example, `{"key": {"expression": "key.encode('utf-8')", "output_type": "bytes"}}`.
+* **projectId**: The Google Cloud project ID of the Bigtable instance. For example, `my-gcp-project`.
+* **instanceId**: The Bigtable instance ID. For example, `my-bigtable-instance`.
+* **tableId**: Bigtable table ID to write the output to. For example, `my-bigtable-table`.
+* **windowing**: Windowing options - see https://beam.apache.org/documentation/sdks/yaml/#windowing For example, `{"type": "fixed", "size": "10s"}`.
+
+### 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.
+* **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`.
+
+
+
+## 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/KafkaToBigtableYaml.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=<my-project>
+export BUCKET_NAME=<bucket-name>
+export ARTIFACT_REGISTRY_REPO=<region>-docker.pkg.dev/$PROJECT/<repo>
+
+mvn clean package -PtemplatesStage  \
+-DskipTests \
+-DprojectId="$PROJECT" \
+-DbucketName="$BUCKET_NAME" \
+-DartifactRegistry="$ARTIFACT_REGISTRY_REPO" \
+-DstagePrefix="templates" \
+-DtemplateName="Kafka_To_Bigtable_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/<project>`.
+
+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://<bucket-name>/templates/flex/Kafka_To_Bigtable_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=<my-project>
+export BUCKET_NAME=<bucket-name>
+export REGION=us-central1
+export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/Kafka_To_Bigtable_Yaml"
+
+### Required
+export BOOTSTRAP_SERVERS=<bootstrapServers>
+export TOPIC=<topic>
+export FIELDS=<fields>
+export PROJECT_ID=<projectId>
+export INSTANCE_ID=<instanceId>
+export TABLE_ID=<tableId>
+export WINDOWING=<windowing>
+
+### Optional
+export CONFLUENT_SCHEMA_REGISTRY_SUBJECT=<confluentSchemaRegistrySubject>
+export CONFLUENT_SCHEMA_REGISTRY_URL=<confluentSchemaRegistryUrl>
+export CONSUMER_CONFIG_UPDATES=<consumerConfigUpdates>
+export FILE_DESCRIPTOR_PATH=<fileDescriptorPath>
+export FORMAT=JSON
+export MESSAGE_NAME=<messageName>
+export SCHEMA=<schema>
+export LANGUAGE=generic
+
+gcloud dataflow flex-template run "kafka-to-bigtable-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 "messageName=$MESSAGE_NAME" \
+  --parameters "schema=$SCHEMA" \
+  --parameters "language=$LANGUAGE" \
+  --parameters "fields=$FIELDS" \
+  --parameters "projectId=$PROJECT_ID" \
+  --parameters "instanceId=$INSTANCE_ID" \
+  --parameters "tableId=$TABLE_ID" \
+  --parameters "windowing=$WINDOWING"
+```
+
+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=<my-project>
+export BUCKET_NAME=<bucket-name>
+export REGION=us-central1
+
+### Required
+export BOOTSTRAP_SERVERS=<bootstrapServers>
+export TOPIC=<topic>
+export FIELDS=<fields>
+export PROJECT_ID=<projectId>
+export INSTANCE_ID=<instanceId>
+export TABLE_ID=<tableId>
+export WINDOWING=<windowing>
+
+### Optional
+export CONFLUENT_SCHEMA_REGISTRY_SUBJECT=<confluentSchemaRegistrySubject>
+export CONFLUENT_SCHEMA_REGISTRY_URL=<confluentSchemaRegistryUrl>
+export CONSUMER_CONFIG_UPDATES=<consumerConfigUpdates>
+export FILE_DESCRIPTOR_PATH=<fileDescriptorPath>
+export FORMAT=JSON
+export MESSAGE_NAME=<messageName>
+export SCHEMA=<schema>
+export LANGUAGE=generic
+
+mvn clean package -PtemplatesRun \
+-DskipTests \
+-DprojectId="$PROJECT" \
+-DbucketName="$BUCKET_NAME" \
+-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" \
+-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_Bigtable_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 = "<my-project>"
+}
+variable "region" {
+  default = "us-central1"
+}
+
+resource "google_dataflow_flex_template_job" "kafka_to_bigtable_yaml" {
+
+  provider          = google-beta
+  container_spec_gcs_path = "gs://dataflow-templates-${var.region}/latest/flex/Kafka_To_Bigtable_Yaml"
+  name              = "kafka-to-bigtable-yaml"
+  region            = var.region
+  parameters        = {
+    bootstrapServers = "<bootstrapServers>"
+    topic = "<topic>"
+    fields = "<fields>"
+    projectId = "<projectId>"
+    instanceId = "<instanceId>"
+    tableId = "<tableId>"
+    windowing = "<windowing>"
+    # confluentSchemaRegistrySubject = "<confluentSchemaRegistrySubject>"
+    # confluentSchemaRegistryUrl = "<confluentSchemaRegistryUrl>"
+    # consumerConfigUpdates = "<consumerConfigUpdates>"
+    # fileDescriptorPath = "<fileDescriptorPath>"
+    # format = "JSON"
+    # messageName = "<messageName>"
+    # schema = "<schema>"
+    # language = "generic"
+  }
+}
+```

yaml/README_Kafka_To_Iceberg_Yaml.md

@@ -23,17 +23,12 @@ 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.
 * **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 +138,12 @@ export CATALOG_PROPERTIES=<catalogProperties>
 export TRIGGERING_FREQUENCY_SECONDS=<triggeringFrequencySeconds>
 
 ### Optional
-export ALLOW_DUPLICATES=<allowDuplicates>
 export CONFLUENT_SCHEMA_REGISTRY_SUBJECT=<confluentSchemaRegistrySubject>
 export CONFLUENT_SCHEMA_REGISTRY_URL=<confluentSchemaRegistryUrl>
 export CONSUMER_CONFIG_UPDATES=<consumerConfigUpdates>
 export FILE_DESCRIPTOR_PATH=<fileDescriptorPath>
 export FORMAT=JSON
 export MESSAGE_NAME=<messageName>
-export OFFSET_DEDUPLICATION=<offsetDeduplication>
-export REDISTRIBUTE_BY_RECORD_KEY=<redistributeByRecordKey>
-export REDISTRIBUTE_NUM_KEYS=<redistributeNumKeys>
-export REDISTRIBUTED=<redistributed>
 export SCHEMA=<schema>
 export CONFIG_PROPERTIES=<configProperties>
 export DROP=<drop>
@@ -170,17 +160,12 @@ 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 "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 +205,12 @@ export CATALOG_PROPERTIES=<catalogProperties>
 export TRIGGERING_FREQUENCY_SECONDS=<triggeringFrequencySeconds>
 
 ### Optional
-export ALLOW_DUPLICATES=<allowDuplicates>
 export CONFLUENT_SCHEMA_REGISTRY_SUBJECT=<confluentSchemaRegistrySubject>
 export CONFLUENT_SCHEMA_REGISTRY_URL=<confluentSchemaRegistryUrl>
 export CONSUMER_CONFIG_UPDATES=<consumerConfigUpdates>
 export FILE_DESCRIPTOR_PATH=<fileDescriptorPath>
 export FORMAT=JSON
 export MESSAGE_NAME=<messageName>
-export OFFSET_DEDUPLICATION=<offsetDeduplication>
-export REDISTRIBUTE_BY_RECORD_KEY=<redistributeByRecordKey>
-export REDISTRIBUTE_NUM_KEYS=<redistributeNumKeys>
-export REDISTRIBUTED=<redistributed>
 export SCHEMA=<schema>
 export CONFIG_PROPERTIES=<configProperties>
 export DROP=<drop>
@@ -248,7 +228,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,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 +279,12 @@ resource "google_dataflow_flex_template_job" "kafka_to_iceberg_yaml" {
     catalogName = "<catalogName>"
     catalogProperties = "<catalogProperties>"
     triggeringFrequencySeconds = "<triggeringFrequencySeconds>"
-    # allowDuplicates = "<allowDuplicates>"
     # confluentSchemaRegistrySubject = "<confluentSchemaRegistrySubject>"
     # confluentSchemaRegistryUrl = "<confluentSchemaRegistryUrl>"
     # consumerConfigUpdates = "<consumerConfigUpdates>"
     # fileDescriptorPath = "<fileDescriptorPath>"
     # format = "JSON"
     # messageName = "<messageName>"
-    # offsetDeduplication = "<offsetDeduplication>"
-    # redistributeByRecordKey = "<redistributeByRecordKey>"
-    # redistributeNumKeys = "<redistributeNumKeys>"
-    # redistributed = "<redistributed>"
     # schema = "<schema>"
     # configProperties = "<configProperties>"
     # drop = "<drop>"