diff --git a/yaml/README_Kafka_To_Bigtable_Yaml.md b/yaml/README_Kafka_To_Bigtable_Yaml.md new file mode 100644 index 0000000000..e00d2839fb --- /dev/null +++ b/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= +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_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/`. + +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_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= +export BUCKET_NAME= +export REGION=us-central1 +export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/Kafka_To_Bigtable_Yaml" + +### Required +export BOOTSTRAP_SERVERS= +export TOPIC= +export FIELDS= +export PROJECT_ID= +export INSTANCE_ID= +export TABLE_ID= +export WINDOWING= + +### Optional +export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= +export CONFLUENT_SCHEMA_REGISTRY_URL= +export CONSUMER_CONFIG_UPDATES= +export FILE_DESCRIPTOR_PATH= +export FORMAT=JSON +export MESSAGE_NAME= +export 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= +export BUCKET_NAME= +export REGION=us-central1 + +### Required +export BOOTSTRAP_SERVERS= +export TOPIC= +export FIELDS= +export PROJECT_ID= +export INSTANCE_ID= +export TABLE_ID= +export WINDOWING= + +### Optional +export CONFLUENT_SCHEMA_REGISTRY_SUBJECT= +export CONFLUENT_SCHEMA_REGISTRY_URL= +export CONSUMER_CONFIG_UPDATES= +export FILE_DESCRIPTOR_PATH= +export FORMAT=JSON +export MESSAGE_NAME= +export 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 = "" +} +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 = "" + topic = "" + fields = "" + projectId = "" + instanceId = "" + tableId = "" + windowing = "" + # confluentSchemaRegistrySubject = "" + # confluentSchemaRegistryUrl = "" + # consumerConfigUpdates = "" + # fileDescriptorPath = "" + # format = "JSON" + # messageName = "" + # schema = "" + # language = "generic" + } +} +``` diff --git a/yaml/README_Kafka_To_Iceberg_Yaml.md b/yaml/README_Kafka_To_Iceberg_Yaml.md index 9346638a26..375cc95dbc 100644 --- a/yaml/README_Kafka_To_Iceberg_Yaml.md +++ b/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= 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 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 +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= 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 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 +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 = "" catalogProperties = "" triggeringFrequencySeconds = "" - # allowDuplicates = "" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" # 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..3411357e9b 100644 --- a/yaml/README_Kafka_to_BigQuery_Yaml.md +++ b/yaml/README_Kafka_to_BigQuery_Yaml.md @@ -28,17 +28,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"}]}`. * **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 +136,12 @@ 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 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 +153,12 @@ 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 "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 +189,12 @@ 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 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 +207,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,messageName=$MESSAGE_NAME,schema=$SCHEMA,table=$TABLE,createDisposition=$CREATE_DISPOSITION,writeDisposition=$WRITE_DISPOSITION,numStreams=$NUM_STREAMS,outputDeadletterTable=$OUTPUT_DEADLETTER_TABLE" \ -f yaml ``` @@ -276,17 +256,12 @@ resource "google_dataflow_flex_template_job" "kafka_to_bigquery_yaml" { topic = "" table = "
" outputDeadletterTable = "" - # allowDuplicates = "" # confluentSchemaRegistrySubject = "" # confluentSchemaRegistryUrl = "" # consumerConfigUpdates = "" # fileDescriptorPath = "" # format = "JSON" # 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/README_PubSub_To_Bigtable_Yaml.md b/yaml/README_PubSub_To_Bigtable_Yaml.md index 07d3c1e49d..d2642ca8dc 100644 --- a/yaml/README_PubSub_To_Bigtable_Yaml.md +++ b/yaml/README_PubSub_To_Bigtable_Yaml.md @@ -22,11 +22,11 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat * **topic**: Pub/Sub topic to read the input from. For example, `projects/your-project-id/topics/your-topic-name`. * **format**: The message format. One of: AVRO, JSON, PROTO, RAW, or STRING. * **schema**: A schema is required if data format is JSON, AVRO or PROTO. For JSON, this is a JSON schema. For AVRO and PROTO, this is the full schema definition. -* **language**: The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic. -* **fields**: The output fields to compute, each mapping to the expression or callable that creates them. -* **projectId**: The Google Cloud project ID of the Bigtable instance. -* **instanceId**: The Bigtable instance ID. -* **tableId**: Bigtable table ID to write the output to. +* **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"}`. * **outputDeadLetterPubSubTopic**: Pub/Sub error topic for failed transformation messages. For example, `projects/your-project-id/topics/your-error-topic-name`. ### Optional parameters @@ -37,7 +37,7 @@ on [Metadata Annotations](https://github.com/GoogleCloudPlatform/DataflowTemplat * **timestampAttribute**: Message value to use as element timestamp. If None, uses message publishing time as the timestamp. Timestamp values should be in one of two formats: 1). A numerical value representing the number of milliseconds since the Unix epoch. 2). A string in RFC 3339 format, UTC timezone. Example: ``2015-10-29T23:41:41.123Z``. The sub-second component of the timestamp is optional, and digits beyond the first three (i.e., time units smaller than milliseconds) may be ignored. * **errorHandling**: This option specifies whether and where to output error rows. * **subscription**: Pub/Sub subscription to read the input from. For example, `projects/your-project-id/subscriptions/your-subscription-name`. -* **windowing**: Windowing options - see https://beam.apache.org/documentation/sdks/yaml/#windowing. +* **language**: The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic. For example, `python`. @@ -133,11 +133,11 @@ export TEMPLATE_SPEC_GCSPATH="gs://$BUCKET_NAME/templates/flex/PubSub_To_Bigtabl export TOPIC= export FORMAT= export SCHEMA= -export LANGUAGE= export FIELDS= export PROJECT_ID= export INSTANCE_ID= export TABLE_ID= +export WINDOWING= export OUTPUT_DEAD_LETTER_PUB_SUB_TOPIC= ### Optional @@ -147,7 +147,7 @@ export ID_ATTRIBUTE= export TIMESTAMP_ATTRIBUTE= export ERROR_HANDLING= export SUBSCRIPTION= -export WINDOWING= +export LANGUAGE=generic gcloud dataflow flex-template run "pubsub-to-bigtable-yaml-job" \ --project "$PROJECT" \ @@ -190,11 +190,11 @@ export REGION=us-central1 export TOPIC= export FORMAT= export SCHEMA= -export LANGUAGE= export FIELDS= export PROJECT_ID= export INSTANCE_ID= export TABLE_ID= +export WINDOWING= export OUTPUT_DEAD_LETTER_PUB_SUB_TOPIC= ### Optional @@ -204,7 +204,7 @@ export ID_ATTRIBUTE= export TIMESTAMP_ATTRIBUTE= export ERROR_HANDLING= export SUBSCRIPTION= -export WINDOWING= +export LANGUAGE=generic mvn clean package -PtemplatesRun \ -DskipTests \ @@ -261,11 +261,11 @@ resource "google_dataflow_flex_template_job" "pubsub_to_bigtable_yaml" { topic = "" format = "" schema = "" - language = "" fields = "" projectId = "" instanceId = "" tableId = "" + windowing = "" outputDeadLetterPubSubTopic = "" # attributes = "" # attributesMap = "" @@ -273,7 +273,7 @@ resource "google_dataflow_flex_template_job" "pubsub_to_bigtable_yaml" { # timestampAttribute = "" # errorHandling = "" # subscription = "" - # windowing = "" + # language = "generic" } } ``` 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..7ac936194e 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.", @@ -126,7 +117,7 @@ public interface KafkaToBigQueryYaml { String getFormat(); @TemplateParameter.Text( - order = 9, + order = 8, name = "messageName", optional = true, description = @@ -136,44 +127,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 = 9, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -184,7 +139,7 @@ public interface KafkaToBigQueryYaml { String getSchema(); @TemplateParameter.Text( - order = 15, + order = 10, name = "table", optional = false, description = "BigQuery table", @@ -195,7 +150,7 @@ public interface KafkaToBigQueryYaml { String getTable(); @TemplateParameter.Text( - order = 16, + order = 11, name = "createDisposition", optional = true, description = "How to create", @@ -206,7 +161,7 @@ public interface KafkaToBigQueryYaml { String getCreateDisposition(); @TemplateParameter.Text( - order = 17, + order = 12, name = "writeDisposition", optional = true, description = "How to write", @@ -217,7 +172,7 @@ public interface KafkaToBigQueryYaml { String getWriteDisposition(); @TemplateParameter.Integer( - order = 18, + order = 13, name = "numStreams", optional = true, description = "Number of streams for BigQuery Storage Write API", @@ -228,7 +183,7 @@ public interface KafkaToBigQueryYaml { Integer getNumStreams(); @TemplateParameter.Text( - order = 19, + order = 14, 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 new file mode 100644 index 0000000000..7a241663a6 --- /dev/null +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYaml.java @@ -0,0 +1,203 @@ +/* + * 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_Bigtable_Yaml", + category = TemplateCategory.STREAMING, + type = Template.TemplateType.YAML, + displayName = "Kafka to Bigtable (YAML)", + description = + "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.", + flexContainerName = "pipeline-yaml", + yamlTemplateFile = "KafkaToBigtable.yaml", + filesToCopy = {"main.py", "requirements.txt"}, + documentation = + "https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-bigtable", + 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 Bigtable table must exist." + }, + streaming = true, + hidden = false) +public interface KafkaToBigtableYaml { + + @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.Text( + order = 8, + 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 = 9, + 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 = 10, + name = "language", + optional = true, + description = "Language used to define the expressions.", + helpText = + "The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic.", + example = "python") + @Default.String("generic") + String getLanguage(); + + @TemplateParameter.Text( + order = 11, + name = "fields", + optional = false, + description = "Field mapping configuration", + helpText = + "The output fields to compute, each mapping to the expression or callable that creates them.", + example = "{\"key\": {\"expression\": \"key.encode('utf-8')\", \"output_type\": \"bytes\"}}") + @Validation.Required + String getFields(); + + @TemplateParameter.Text( + order = 12, + name = "projectId", + optional = false, + description = "Bigtable project ID", + helpText = "The Google Cloud project ID of the Bigtable instance.", + example = "my-gcp-project") + @Validation.Required + String getProjectId(); + + @TemplateParameter.Text( + order = 13, + name = "instanceId", + optional = false, + description = "Bigtable instance ID", + helpText = "The Bigtable instance ID.", + example = "my-bigtable-instance") + @Validation.Required + String getInstanceId(); + + @TemplateParameter.Text( + order = 14, + name = "tableId", + optional = false, + description = "Bigtable output table", + helpText = "Bigtable table ID to write the output to.", + example = "my-bigtable-table") + @Validation.Required + String getTableId(); + + @TemplateParameter.Text( + order = 15, + name = "windowing", + optional = false, + description = "Windowing options", + helpText = + "Windowing options - see https://beam.apache.org/documentation/sdks/yaml/#windowing", + example = "{\"type\": \"fixed\", \"size\": \"10s\"}") + @Validation.Required + String getWindowing(); +} 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..de7b718932 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.", @@ -129,7 +120,7 @@ public interface KafkaToIcebergYaml { String getFormat(); @TemplateParameter.Text( - order = 9, + order = 8, name = "messageName", optional = true, description = @@ -139,44 +130,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 = 9, name = "schema", optional = true, description = "The schema in which the data is encoded in the Kafka topic.", @@ -187,7 +142,7 @@ public interface KafkaToIcebergYaml { String getSchema(); @TemplateParameter.Text( - order = 15, + order = 10, name = "table", optional = false, description = "A fully-qualified table identifier.", @@ -197,7 +152,7 @@ public interface KafkaToIcebergYaml { String getTable(); @TemplateParameter.Text( - order = 16, + order = 11, name = "catalogName", optional = false, description = "Name of the catalog containing the table.", @@ -207,7 +162,7 @@ public interface KafkaToIcebergYaml { String getCatalogName(); @TemplateParameter.Text( - order = 17, + order = 12, name = "catalogProperties", optional = false, description = "Properties used to set up the Iceberg catalog.", @@ -217,7 +172,7 @@ public interface KafkaToIcebergYaml { String getCatalogProperties(); @TemplateParameter.Text( - order = 18, + order = 13, name = "configProperties", optional = true, description = "Properties passed to the Hadoop Configuration.", @@ -226,7 +181,7 @@ public interface KafkaToIcebergYaml { String getConfigProperties(); @TemplateParameter.Text( - order = 19, + order = 14, name = "drop", optional = true, description = "A list of field names to drop from the input record before writing.", @@ -235,7 +190,7 @@ public interface KafkaToIcebergYaml { String getDrop(); @TemplateParameter.Text( - order = 20, + order = 15, name = "keep", optional = true, description = "A list of field names to keep in the input record.", @@ -244,7 +199,7 @@ public interface KafkaToIcebergYaml { String getKeep(); @TemplateParameter.Text( - order = 21, + order = 16, name = "only", optional = true, description = "The name of a single record field that should be written.", @@ -253,7 +208,7 @@ public interface KafkaToIcebergYaml { String getOnly(); @TemplateParameter.Text( - order = 22, + order = 17, name = "partitionFields", optional = true, description = "Fields used to create a partition spec for new tables.", @@ -262,7 +217,7 @@ public interface KafkaToIcebergYaml { String getPartitionFields(); @TemplateParameter.Text( - order = 23, + order = 18, name = "tableProperties", optional = true, description = "Iceberg table properties to be set on table creation.", @@ -271,7 +226,7 @@ public interface KafkaToIcebergYaml { String getTableProperties(); @TemplateParameter.Integer( - order = 24, + order = 19, name = "triggeringFrequencySeconds", optional = false, description = "For a streaming pipeline, the frequency at which snapshots are produced.", @@ -281,7 +236,7 @@ public interface KafkaToIcebergYaml { Integer getTriggeringFrequencySeconds(); @TemplateParameter.Text( - order = 25, + order = 20, name = "sdfCheckpointAfterDuration", optional = true, description = "Dataflow Pipeline Option: Duration after which to checkpoint stateful DoFns.", @@ -292,7 +247,7 @@ public interface KafkaToIcebergYaml { String getSdfCheckpointAfterDuration(); @TemplateParameter.Integer( - order = 26, + order = 21, name = "sdfCheckpointAfterOutputBytes", optional = true, description = diff --git a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigtableYaml.java b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigtableYaml.java index 7f052eb1e0..1e859c20eb 100644 --- a/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigtableYaml.java +++ b/yaml/src/main/java/com/google/cloud/teleport/templates/yaml/PubSubToBigtableYaml.java @@ -18,6 +18,7 @@ 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( @@ -135,12 +136,12 @@ public interface PubSubToBigtableYaml { @TemplateParameter.Text( order = 10, name = "language", - optional = false, + optional = true, description = "Language used to define the expressions.", helpText = "The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic.", - example = "") - @Validation.Required + example = "python") + @Default.String("generic") String getLanguage(); @TemplateParameter.Text( @@ -150,7 +151,7 @@ public interface PubSubToBigtableYaml { description = "Field mapping configuration", helpText = "The output fields to compute, each mapping to the expression or callable that creates them.", - example = "") + example = "{\"key\": {\"expression\": \"key.encode('utf-8')\", \"output_type\": \"bytes\"}}") @Validation.Required String getFields(); @@ -160,7 +161,7 @@ public interface PubSubToBigtableYaml { optional = false, description = "Bigtable project ID", helpText = "The Google Cloud project ID of the Bigtable instance.", - example = "") + example = "my-gcp-project") @Validation.Required String getProjectId(); @@ -170,7 +171,7 @@ public interface PubSubToBigtableYaml { optional = false, description = "Bigtable instance ID", helpText = "The Bigtable instance ID.", - example = "") + example = "my-bigtable-instance") @Validation.Required String getInstanceId(); @@ -180,18 +181,19 @@ public interface PubSubToBigtableYaml { optional = false, description = "Bigtable output table", helpText = "Bigtable table ID to write the output to.", - example = "") + example = "my-bigtable-table") @Validation.Required String getTableId(); @TemplateParameter.Text( order = 15, name = "windowing", - optional = true, + optional = false, description = "Windowing options", helpText = "Windowing options - see https://beam.apache.org/documentation/sdks/yaml/#windowing", - example = "") + example = "{\"type\": \"fixed\", \"size\": \"10s\"}") + @Validation.Required String getWindowing(); @TemplateParameter.Text( diff --git a/yaml/src/main/python/options/bigtable_options.yaml b/yaml/src/main/python/options/bigtable_options.yaml index aa9d659b6d..d49e1a9bd4 100644 --- a/yaml/src/main/python/options/bigtable_options.yaml +++ b/yaml/src/main/python/options/bigtable_options.yaml @@ -5,18 +5,21 @@ options: name: "projectId" description: "Bigtable project ID" help: "The Google Cloud project ID of the Bigtable instance." + example: "my-gcp-project" required: true type: text - order: 2 name: "instanceId" description: "Bigtable instance ID" help: "The Bigtable instance ID." + example: "my-bigtable-instance" required: true type: text - order: 3 name: "tableId" description: "Bigtable output table" help: "Bigtable table ID to write the output to." + example: "my-bigtable-table" required: true type: text @@ -26,6 +29,7 @@ options: name: "flatten" description: "Bigtable flatten boolean" help: "If set to false, output rows are nested; if true or omitted, output rows are flattened." + example: true default: true required: false type: boolean diff --git a/yaml/src/main/python/options/kafka_options.yaml b/yaml/src/main/python/options/kafka_options.yaml index d64c4c4e49..1ec99aae67 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,42 +51,14 @@ options: required: false default: "JSON" type: text - - order: 9 + - order: 8 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 - 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 + - order: 9 name: "schema" description: "The schema in which the data is encoded in the Kafka topic." help: > diff --git a/yaml/src/main/python/options/maptofields_options.yaml b/yaml/src/main/python/options/maptofields_options.yaml index f50d050eda..bcdeeadc36 100644 --- a/yaml/src/main/python/options/maptofields_options.yaml +++ b/yaml/src/main/python/options/maptofields_options.yaml @@ -7,7 +7,9 @@ options: help: > The language used to define (and execute) the expressions and/or callables in fields. Defaults to generic. - required: true + example: "python" + default: "generic" + required: false type: text - order: 2 name: "fields" @@ -15,6 +17,7 @@ options: help: > The output fields to compute, each mapping to the expression or callable that creates them. + example: "{\"key\": {\"expression\": \"key.encode('utf-8')\", \"output_type\": \"bytes\"}}" required: true type: text diff --git a/yaml/src/main/python/options/windowinto_options.yaml b/yaml/src/main/python/options/windowinto_options.yaml index 461405d0c4..b0f54332b8 100644 --- a/yaml/src/main/python/options/windowinto_options.yaml +++ b/yaml/src/main/python/options/windowinto_options.yaml @@ -5,6 +5,7 @@ options: name: "windowing" description: "Windowing options" help: "Windowing options - see https://beam.apache.org/documentation/sdks/yaml/#windowing" - is_optional: false + example: '{"type": "fixed", "size": "10s"}' + required: true type: text diff --git a/yaml/src/main/yaml/KafkaToBigtable.yaml b/yaml/src/main/yaml/KafkaToBigtable.yaml new file mode 100644 index 0000000000..0dceaa22f7 --- /dev/null +++ b/yaml/src/main/yaml/KafkaToBigtable.yaml @@ -0,0 +1,78 @@ +template: + + name: "Kafka_To_Bigtable_Yaml" + category: "STREAMING" + type: "YAML" + display_name: "Kafka to Bigtable (YAML)" + description: > + 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. + flex_container_name: "pipeline-yaml" + yamlTemplateFile: "KafkaToBigtable.yaml" + filesToCopy: > + {"main.py", "requirements.txt"} + documentation: > + https://cloud.google.com/dataflow/docs/guides/templates/provided-yaml/kafka-to-bigtable + 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 Bigtable table must exist." + } + streaming: true + hidden: false + + options_file: + - "kafka_options" + - "maptofields_options" + - "windowinto_options" + - "bigtable_options" + + parameters: + - kafka_read_options + - maptofields_common_options + - bigtable_common_options + - windowinto_common_options + + +pipeline: + type: composite + 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') }} + message_name: {{ messageName }} + {% if schema %} + schema: | + {{ schema }} + {% endif %} + auto_offset_reset_config: 'earliest' + - type: MapToFields + name: ConvertStringsToBytes + input: ReadFromKafka + config: + language: {{ language }} + fields: {{ fields }} + - type: WindowInto + name: WindowInto + input: ConvertStringsToBytes + config: + windowing: {{ windowing }} + - type: WriteToBigTable + name: WriteToBigtable + input: WindowInto + config: + project: {{ projectId }} + instance: {{ instanceId }} + table: {{ tableId }} + +options: + streaming: true diff --git a/yaml/src/main/yaml/KafkaToIceberg.yaml b/yaml/src/main/yaml/KafkaToIceberg.yaml index 3be059038c..749a4ceee3 100644 --- a/yaml/src/main/yaml/KafkaToIceberg.yaml +++ b/yaml/src/main/yaml/KafkaToIceberg.yaml @@ -52,17 +52,12 @@ pipeline: config: bootstrap_servers: {{ bootstrapServers }} topic: {{ topic }} - allow_duplicates: {{ allowDuplicates }} confluent_schema_registry_subject: {{ confluentSchemaRegistrySubject }} confluent_schema_registry_url: {{ confluentSchemaRegistryUrl }} - consumer_config_updates: {{ consumerConfigUpdates }} + consumer_config: {{ consumerConfigUpdates }} file_descriptor_path: {{ fileDescriptorPath }} format: {{ format | default('JSON') }} message_name: {{ messageName }} - offset_deduplication: {{ offsetDeduplication }} - redistribute_by_record_key: {{ redistributeByRecordKey }} - redistribute_num_keys: {{ redistributeNumKeys }} - redistributed: {{ redistributed }} {% if schema %} schema: | {{ schema }} diff --git a/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYamlIT.java b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYamlIT.java new file mode 100644 index 0000000000..b9f275f3d2 --- /dev/null +++ b/yaml/src/test/java/com/google/cloud/teleport/templates/yaml/KafkaToBigtableYamlIT.java @@ -0,0 +1,224 @@ +/* + * 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.bigtable.BigtableResourceManagerUtils.generateTableId; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatPipeline; +import static org.apache.beam.it.truthmatchers.PipelineAsserts.assertThatResult; + +import com.google.cloud.bigtable.data.v2.models.Row; +import com.google.cloud.teleport.metadata.SkipDirectRunnerTest; +import com.google.cloud.teleport.metadata.TemplateIntegrationTest; +import com.google.common.collect.ImmutableList; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.function.Function; +import java.util.stream.Collectors; +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.bigtable.BigtableResourceManager; +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 KafkaToBigtableYaml}. */ +@Category({TemplateIntegrationTest.class, SkipDirectRunnerTest.class}) +@TemplateIntegrationTest(KafkaToBigtableYaml.class) +@RunWith(JUnit4.class) +public final class KafkaToBigtableYamlIT extends TemplateTestBase { + + private static final Logger LOG = LoggerFactory.getLogger(KafkaToBigtableYamlIT.class); + + private KafkaResourceManager kafkaResourceManager; + private BigtableResourceManager bigtableResourceManager; + + @Before + public void setup() throws IOException { + kafkaResourceManager = + KafkaResourceManager.builder(testName).setHost(TestProperties.hostIp()).build(); + bigtableResourceManager = + BigtableResourceManager.builder(testName, PROJECT, credentialsProvider) + .maybeUseStaticInstance() + .build(); + } + + @After + public void tearDown() { + ResourceManagerUtils.cleanResources(kafkaResourceManager, bigtableResourceManager); + } + + @Test + public void testKafkaToBigtable() throws IOException { + kafkaToBigtable(Function.identity()); + } + + public void kafkaToBigtable( + Function + paramsAdder) + throws IOException { + + LOG.info("Starting kafkaToBigtable test. Test name: {}. Spec path: {}", testName, specPath); + + /******************************* Arrange ********************************/ + + LOG.info("Creating Kafka topic..."); + String topicName = kafkaResourceManager.createTopic(testName, 5); + + LOG.info("Creating Bigtable table..."); + String tableId = generateTableId("test_table"); + bigtableResourceManager.createTable(tableId, ImmutableList.of("cf1")); + + 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\":{\"key\":{\"type\":\"string\"},\"type\":{\"type\":\"string\"},\"family_name\":{\"type\":\"string\"},\"column_qualifier\":{\"type\":\"string\"},\"value\":{\"type\":\"string\"},\"timestamp_micros\":{\"type\":\"integer\"}}}") + .addParameter("windowing", "{\"type\":\"fixed\",\"size\":\"10s\"}") + .addParameter("tableId", tableId) + .addParameter("instanceId", bigtableResourceManager.getInstanceId()) + .addParameter("projectId", PROJECT) + .addParameter("language", "python") + .addParameter( + "fields", + "{" + + "\"key\": {\"expression\": \"key.encode('utf-8')\", \"output_type\": \"bytes\"}," + + "\"type\": {\"expression\": \"type\", \"output_type\": \"string\"}," + + "\"family_name\": {\"expression\": \"family_name\", \"output_type\": \"string\"}," + + "\"column_qualifier\": {\"expression\": \"column_qualifier.encode('utf-8')\", \"output_type\": \"bytes\"}," + + "\"value\": {\"expression\": \"value.encode('utf-8')\", \"output_type\": \"bytes\"}," + + "\"timestamp_micros\": {\"expression\": \"timestamp_micros\", \"output_type\": \"integer\"}" + + "}")); + + /********************************* 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..."); + 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), + "{\"key\": \"row" + + id1 + + "\", \"type\": \"SetCell\", \"family_name\": \"cf1\", \"column_qualifier\": \"cq1\", \"value\": \"value1\", \"timestamp_micros\": 5000}"); + publish( + kafkaProducer, + topicName, + String.valueOf(id2), + "{\"key\": \"row" + + id2 + + "\", \"type\": \"SetCell\", \"family_name\": \"cf1\", \"column_qualifier\": \"cq2\", \"value\": \"value2\", \"timestamp_micros\": 1000}"); + } + } + + LOG.info("Waiting for pipeline condition..."); + PipelineOperator.Result result = + pipelineOperator() + .waitForConditionAndFinish( + createConfig(info), + () -> { + List rows = bigtableResourceManager.readTable(tableId); + if (rows == null) { + LOG.warn("bigtableResourceManager.readTable(tableId) returned null. Retrying."); + return false; + } + int tableSize = rows.size(); + LOG.info("Checking table size. Current size: {}", tableSize); + return tableSize == 20; + }); + + /******************************** Assert ********************************/ + assertThatResult(result).meetsConditions(); + + LOG.info("Verifying 20 rows in the Bigtable still exist..."); + List tableRows = bigtableResourceManager.readTable(tableId); + assertThat(tableRows).hasSize(20); + + LOG.info("Verifying the exact 20 rows in Bigtable..."); + Map rowMap = + tableRows.stream() + .collect(Collectors.toMap(row -> row.getKey().toStringUtf8(), row -> row)); + + for (int i = 1; i <= 10; i++) { + String key1 = "row" + i + "1"; + assertThat(rowMap).containsKey(key1); + Row row1 = rowMap.get(key1); + assertThat(row1.getCells()).hasSize(1); + assertThat(row1.getCells().get(0).getFamily()).isEqualTo("cf1"); + assertThat(row1.getCells().get(0).getQualifier().toStringUtf8()).isEqualTo("cq1"); + assertThat(row1.getCells().get(0).getValue().toStringUtf8()).isEqualTo("value1"); + assertThat(row1.getCells().get(0).getTimestamp()).isEqualTo(5000L); + + String key2 = "row" + i + "2"; + assertThat(rowMap).containsKey(key2); + Row row2 = rowMap.get(key2); + assertThat(row2.getCells()).hasSize(1); + assertThat(row2.getCells().get(0).getFamily()).isEqualTo("cf1"); + assertThat(row2.getCells().get(0).getQualifier().toStringUtf8()).isEqualTo("cq2"); + assertThat(row2.getCells().get(0).getValue().toStringUtf8()).isEqualTo("value2"); + assertThat(row2.getCells().get(0).getTimestamp()).isEqualTo(1000L); + } + } + + 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); + } + } +}