From 7d82725508a09389a47c5b1a0377eca220f7d851 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Tue, 16 Dec 2025 18:43:31 +0000 Subject: [PATCH 01/15] Prompt to Gemini : Create a PySpark script to tranform data in GCS from parquet to avro and use the add_insertion_time_column function in @data_tranformer.py to add an additional column Code Generated by Gemini for GCS to GCS Successfully ran serverless Spark job using following command: gcloud dataproc batches submit pyspark transform_parquet_to_avro.py --batch="parquet-to-avro-$(date +%s)" \ --jars=file:///usr/lib/spark/connector/spark-avro.jar --py-files=./data_transformer.py \ --deps-bucket=gs://dataproc-templates-python-deps \ -- --input=gs://dataproc-templates_cloudbuild/gemini-codegen/transform_parquet_to_avro/input/parquet-table \ --output=gs://dataproc-templates_cloudbuild/gemini-codegen/transform_parquet_to_avro/output/avro_table Confirmed by using BigQuery, creating parquet and avro files as external tables. --- .gitignore | 5 +- gemini-codegen/python/data_transformer.py | 48 +++++++++++++++++++ .../python/transform_parquet_to_avro.py | 39 +++++++++++++++ .../python/transform_parquet_to_avroREADME.md | 28 +++++++++++ 4 files changed, 119 insertions(+), 1 deletion(-) create mode 100644 gemini-codegen/python/data_transformer.py create mode 100644 gemini-codegen/python/transform_parquet_to_avro.py create mode 100644 gemini-codegen/python/transform_parquet_to_avroREADME.md diff --git a/.gitignore b/.gitignore index b0a01ff18..08df61aff 100644 --- a/.gitignore +++ b/.gitignore @@ -28,4 +28,7 @@ metastore_db/ python/elastic.jks python/tls.cert -python/venv** \ No newline at end of file +python/venv** + +# Gemini Codegen +gemini-codegen/python/env/ diff --git a/gemini-codegen/python/data_transformer.py b/gemini-codegen/python/data_transformer.py new file mode 100644 index 000000000..87a6bf2dc --- /dev/null +++ b/gemini-codegen/python/data_transformer.py @@ -0,0 +1,48 @@ +from pyspark.sql import DataFrame +from pyspark.sql.functions import current_timestamp + +def add_insertion_time_column(df: DataFrame) -> DataFrame: + """ + Adds a new column 'insertion_time' to the PySpark DataFrame with the current timestamp. + + Args: + df: The input PySpark DataFrame. + + Returns: + A new PySpark DataFrame with the 'insertion_time' column added. + """ + return df.withColumn("insertion_time", current_timestamp()) + +if __name__ == "__main__": + # This is a simple example to demonstrate the function. + # In a real scenario, you would have an active SparkSession. + try: + from pyspark.sql import SparkSession + from pyspark.sql.types import StructType, StructField, StringType, IntegerType + + spark = SparkSession.builder.appName("TimestampColumnExample").getOrCreate() + + # Create a sample DataFrame + schema = StructType([ + StructField("name", StringType(), True), + StructField("age", IntegerType(), True) + ]) + data = [("Alice", 1), ("Bob", 2), ("Charlie", 3)] + sample_df = spark.createDataFrame(data, schema) + + print("Original DataFrame:") + sample_df.show() + + # Add the insertion time column + df_with_time = add_insertion_time_column(sample_df) + + print("\nDataFrame with insertion_time column:") + df_with_time.show() + df_with_time.printSchema() + + spark.stop() + except ImportError: + print("PySpark is not installed. Please install PySpark to run the example:") + print("pip install pyspark") + except Exception as e: + print(f"An error occurred: {e}") diff --git a/gemini-codegen/python/transform_parquet_to_avro.py b/gemini-codegen/python/transform_parquet_to_avro.py new file mode 100644 index 000000000..cd78c7a64 --- /dev/null +++ b/gemini-codegen/python/transform_parquet_to_avro.py @@ -0,0 +1,39 @@ + +import argparse +from pyspark.sql import SparkSession, DataFrame +from data_transformer import add_insertion_time_column + +def transform_data(spark: SparkSession, input_path: str, output_path: str): + """ + Reads data from a Parquet file, adds an insertion time column, and writes to an Avro file. + + Args: + spark: The SparkSession object. + input_path: The GCS path to the input Parquet data. + output_path: The GCS path to write the output Avro data. + """ + # Read the Parquet data from GCS + input_df = spark.read.parquet(input_path) + + # Add the insertion time column + transformed_df = add_insertion_time_column(input_df) + + # Write the transformed data to GCS in Avro format + # You need to have the spark-avro package. + # Example for spark-submit: + # --packages org.apache.spark:spark-avro_2.12:3.3.0 + transformed_df.write.format("avro").save(output_path) + + print(f"Data successfully transformed and saved to {output_path}") + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="PySpark data transformation script") + parser.add_argument("--input", required=True, help="Input GCS path for Parquet data") + parser.add_argument("--output", required=True, help="Output GCS path for Avro data") + args = parser.parse_args() + + spark = SparkSession.builder.appName("Parquet to Avro Transformation").getOrCreate() + + transform_data(spark, args.input, args.output) + + spark.stop() diff --git a/gemini-codegen/python/transform_parquet_to_avroREADME.md b/gemini-codegen/python/transform_parquet_to_avroREADME.md new file mode 100644 index 000000000..d9198e34b --- /dev/null +++ b/gemini-codegen/python/transform_parquet_to_avroREADME.md @@ -0,0 +1,28 @@ +# Dataproc Templates - PySpark Data Transformation + +This directory contains a PySpark script for transforming data on Google Cloud Dataproc. + +## Scripts + +* `data_transformer.py`: A module that contains functions for data transformation. +* `transform_parquet_to_avro.py`: A PySpark script that reads data in Parquet format from a GCS bucket, adds an `insertion_time` column, and writes the transformed data to another GCS bucket in Avro format. + +## How to run the script + +To run the `transform_parquet_to_avro.py` script, you can use the `gcloud dataproc jobs submit pyspark` command. You will need to provide the input and output GCS paths. + +**Example:** + +```bash +gcloud dataproc jobs submit pyspark transform_parquet_to_avro.py \ + --cluster= \ + --region= \ + --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.23.2.jar \ + -- \ + --input=gs:///path/to/parquet \ + --output=gs:///path/to/avro +``` + +**Note:** You need to have the `spark-avro` package available in your cluster. You can add it using the `--packages` option when submitting the job, for example: `--packages org.apache.spark:spark-avro_2.12:3.3.0`. Alternatively, you can install it on your cluster nodes. +You also need to make sure that the `data_transformer.py` file is available to your PySpark job. You can either use `gcloud dataproc jobs submit pyspark --py-files` to ship it with your job, or install it as a package on the cluster. +For more information, see the [Dataproc documentation](https://cloud.google.com/dataproc/docs). From 6fb2f97f42c417b46c049c484799655ccc2db511 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Tue, 16 Dec 2025 20:33:18 +0000 Subject: [PATCH 02/15] Create a PySpark script to extract and transform a hive table, adding an insertion_time column using the add_insertion_time_column function in @data_tranformer.py. Save this table to BigQuery, providing detailed instructions to run this script against a dataproc cluster. Save a summary of this session to hive_to_BQReadme.md Gemini generated Hive to BQ transformation script Comments: spark-bigquery comes preinstalled on dataproc clusters version 2.1 and higher, so no jars need to be provided. The following command worked: gcloud dataproc jobs submit pyspark gs://dataproc-templates_cloudbuild/gemini-codegen/transform_hive_to_bq/src/transform_hive_to_bigquery.py \ --cluster=mixer-test2 --py-files=gs://dataproc-templates_cloudbuild/gemini-codegen/transform_hive_to_bq/src/data_transformer.py \ --properties=spark.hadoop.hive.metastore.uris=thrift://10.115.64.27:9083 \ -- --hive_database=test_db --hive_table=employees --bq_table=gemini_codegen.py_hive_to_bq \ --bq_temp_gcs_bucket=gs://dataproc-templates-python-deps --- gemini-codegen/python/hive_to_BQReadme.md | 44 +++++++++++++++++ .../python/transform_hive_to_bigquery.py | 49 +++++++++++++++++++ 2 files changed, 93 insertions(+) create mode 100644 gemini-codegen/python/hive_to_BQReadme.md create mode 100644 gemini-codegen/python/transform_hive_to_bigquery.py diff --git a/gemini-codegen/python/hive_to_BQReadme.md b/gemini-codegen/python/hive_to_BQReadme.md new file mode 100644 index 000000000..5889f8304 --- /dev/null +++ b/gemini-codegen/python/hive_to_BQReadme.md @@ -0,0 +1,44 @@ +# Dataproc Templates - PySpark Hive to BigQuery + +This directory contains a PySpark script for transforming data from Hive and loading it into BigQuery. + +## Scripts + +* `data_transformer.py`: A module that contains functions for data transformation. +* `transform_hive_to_bigquery.py`: A PySpark script that reads a table from Hive, adds an `insertion_time` column, and writes the transformed data to a BigQuery table. + +## How to run the script + +To run the `transform_hive_to_bigquery.py` script, you can use the `gcloud dataproc jobs submit pyspark` command. You will need to provide the Hive and BigQuery table details. + +**Prerequisites:** + +1. A running Dataproc cluster. +2. The BigQuery Connector for Spark must be available on the cluster. You can add it using the `--jars` flag when submitting the job. +3. The script `transform_hive_to_bigquery.py` and the module `data_transformer.py` must be accessible to the job. The easiest way to do this is to upload them to a GCS bucket and then specify them using the `--py-files` flag. + +**Example:** + +```bash +gcloud dataproc jobs submit pyspark gs:///path/to/transform_hive_to_bigquery.py \ + --cluster= \ + --region= \ + --py-files=gs:///path/to/data_transformer.py \ + --jars=gs://spark-lib/bigquery/spark-bigquery-with-dependencies_2.12-0.23.2.jar \ + -- \ + --hive_database= \ + --hive_table= \ + --bq_table=. \ + --bq_temp_gcs_bucket= +``` + +### **Argument Reference:** + +* `--hive_database`: The name of the source Hive database. +* `--hive_table`: The name of the source Hive table. +* `--bq_table`: The destination BigQuery table in `dataset.table` format. +* `--bq_temp_gcs_bucket`: A GCS bucket used by the BigQuery connector to temporarily store data before writing it to BigQuery. + +## Session Summary + +This session involved creating a PySpark script to extract a Hive table, transform it by adding an `insertion_time` column, and load it into BigQuery. The `add_insertion_time_column` function was reused from the existing `data_transformer.py` module. A README file was also created to provide detailed instructions on how to run this script on a Dataproc cluster, including the necessary `gcloud` command and argument explanations. diff --git a/gemini-codegen/python/transform_hive_to_bigquery.py b/gemini-codegen/python/transform_hive_to_bigquery.py new file mode 100644 index 000000000..675ff137b --- /dev/null +++ b/gemini-codegen/python/transform_hive_to_bigquery.py @@ -0,0 +1,49 @@ + +import argparse +from pyspark.sql import SparkSession +from data_transformer import add_insertion_time_column + +def transform_hive_to_bigquery(spark: SparkSession, hive_database: str, hive_table: str, bq_table: str, bq_temp_gcs_bucket: str): + """ + Reads a Hive table, adds an insertion_time column, and writes to a BigQuery table. + + Args: + spark: The SparkSession object. + hive_database: The name of the source Hive database. + hive_table: The name of the source Hive table. + bq_table: The destination BigQuery table (e.g., 'dataset.table'). + bq_temp_gcs_bucket: The GCS bucket for temporary BigQuery connector data. + """ + # Read data from Hive table + input_df = spark.table(f'{hive_database}.{hive_table}') + + # Add the insertion time column + transformed_df = add_insertion_time_column(input_df) + + # Write the transformed data to BigQuery + transformed_df.write \ + .format('bigquery') \ + .option('table', bq_table) \ + .option('temporaryGcsBucket', bq_temp_gcs_bucket) \ + .mode('append') \ + .save() + + print(f"Data successfully written to BigQuery table: {bq_table}") + +if __name__ == '__main__': + parser = argparse.ArgumentParser(description='PySpark Hive to BigQuery transformation script') + parser.add_argument('--hive_database', required=True, help='Source Hive database') + parser.add_argument('--hive_table', required=True, help='Source Hive table') + parser.add_argument('--bq_table', required=True, help='Destination BigQuery table (dataset.table)') + parser.add_argument('--bq_temp_gcs_bucket', required=True, help='GCS bucket for temporary BigQuery connector data') + args = parser.parse_args() + + # Create a SparkSession with Hive support + spark = SparkSession.builder \ + .appName('Hive to BigQuery Transformation') \ + .enableHiveSupport() \ + .getOrCreate() + + transform_hive_to_bigquery(spark, args.hive_database, args.hive_table, args.bq_table, args.bq_temp_gcs_bucket) + + spark.stop() From 0a850a8e24953a48a4c6acdf72e596b0f1ea0ce5 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Wed, 17 Dec 2025 22:59:39 +0000 Subject: [PATCH 03/15] Final Comments --- gemini-codegen/python/hive_to_BQReadme.md | 14 ++++++++++++++ .../python/transform_parquet_to_avroREADME.md | 16 ++++++++++++++++ 2 files changed, 30 insertions(+) diff --git a/gemini-codegen/python/hive_to_BQReadme.md b/gemini-codegen/python/hive_to_BQReadme.md index 5889f8304..db1fc158c 100644 --- a/gemini-codegen/python/hive_to_BQReadme.md +++ b/gemini-codegen/python/hive_to_BQReadme.md @@ -2,6 +2,20 @@ This directory contains a PySpark script for transforming data from Hive and loading it into BigQuery. +--- +The text between the line above and line below was written by a human. The rest of the document was created by Gemini. The initial prompt to Gemini was: +``` + Create a PySpark script to extract and transform a hive table, adding an insertion_time column using the add_insertion_time_column function in @data_tranformer.py. Save this table to BigQuery, providing detailed instructions to run this script against a dataproc cluster. Save a summary of this session to hive_to_BQReadme.md +``` +Gemini generated the Pyspark script, specifically the file `transform_hive_to_bigquery.py` and the README file. Minor changes were required to run the script because a) the preinstalled Spark BigQuery connector in cluster versions 2.1 and higher was used and b) the Dataproc Hive Metastore service with the thrift protocol was used. The working script is +``` +gcloud dataproc jobs submit pyspark gs://path_to_src/transform_hive_to_bigquery.py \ + --cluster= --py-files=gs://path_to_src/data_transformer.py \ + --properties=spark.hadoop.hive.metastore.uris= \ + -- --hive_database= --hive_table= --bq_table=.
\ + --bq_temp_gcs_bucket= +``` +--- ## Scripts * `data_transformer.py`: A module that contains functions for data transformation. diff --git a/gemini-codegen/python/transform_parquet_to_avroREADME.md b/gemini-codegen/python/transform_parquet_to_avroREADME.md index d9198e34b..d8ef95e71 100644 --- a/gemini-codegen/python/transform_parquet_to_avroREADME.md +++ b/gemini-codegen/python/transform_parquet_to_avroREADME.md @@ -2,6 +2,22 @@ This directory contains a PySpark script for transforming data on Google Cloud Dataproc. +--- +The text between the line above and line below was written by a human. The rest of the document was created by Gemini. The initial prompt to Gemini was: +``` + Create a PySpark script to tranform data in GCS from parquet to avro and use the add_insertion_time_column function in @data_tranformer.py to add an additional column +``` +Gemini generated the Pyspark script, specifically the file `transform_parquet_to_avro.py` and the README file. Minor changes were required to run the script. The working script is +``` + gcloud dataproc batches submit pyspark transform_parquet_to_avro.py \ + --batch="parquet-to-avro-$(date +%s)" \ + --jars=file:///usr/lib/spark/connector/spark-avro.jar \ + --py-files=./data_transformer.py \ + --deps-bucket=gs://temporary_bucket_name \ + -- --input=gs://path_to_input_table/parquet-table \ + --output=gs://path_to_output_table/avro_table +``` +--- ## Scripts * `data_transformer.py`: A module that contains functions for data transformation. From ae56a230fad95e924b10cf38364fed2d7056c0aa Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Wed, 17 Dec 2025 23:19:27 +0000 Subject: [PATCH 04/15] Minor syntax update --- gemini-codegen/python/hive_to_BQReadme.md | 2 +- gemini-codegen/python/transform_parquet_to_avroREADME.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/gemini-codegen/python/hive_to_BQReadme.md b/gemini-codegen/python/hive_to_BQReadme.md index db1fc158c..ee5a0f950 100644 --- a/gemini-codegen/python/hive_to_BQReadme.md +++ b/gemini-codegen/python/hive_to_BQReadme.md @@ -7,7 +7,7 @@ The text between the line above and line below was written by a human. The rest ``` Create a PySpark script to extract and transform a hive table, adding an insertion_time column using the add_insertion_time_column function in @data_tranformer.py. Save this table to BigQuery, providing detailed instructions to run this script against a dataproc cluster. Save a summary of this session to hive_to_BQReadme.md ``` -Gemini generated the Pyspark script, specifically the file `transform_hive_to_bigquery.py` and the README file. Minor changes were required to run the script because a) the preinstalled Spark BigQuery connector in cluster versions 2.1 and higher was used and b) the Dataproc Hive Metastore service with the thrift protocol was used. The working script is +Gemini generated the Pyspark script, specifically the file `transform_hive_to_bigquery.py` and the README file. Minor changes were required to run the script because a) the preinstalled Spark BigQuery connector in cluster versions 2.1 and higher was used and b) the Dataproc Hive Metastore service with the thrift protocol was used. The working gcloud command is ``` gcloud dataproc jobs submit pyspark gs://path_to_src/transform_hive_to_bigquery.py \ --cluster= --py-files=gs://path_to_src/data_transformer.py \ diff --git a/gemini-codegen/python/transform_parquet_to_avroREADME.md b/gemini-codegen/python/transform_parquet_to_avroREADME.md index d8ef95e71..8a097a053 100644 --- a/gemini-codegen/python/transform_parquet_to_avroREADME.md +++ b/gemini-codegen/python/transform_parquet_to_avroREADME.md @@ -7,7 +7,7 @@ The text between the line above and line below was written by a human. The rest ``` Create a PySpark script to tranform data in GCS from parquet to avro and use the add_insertion_time_column function in @data_tranformer.py to add an additional column ``` -Gemini generated the Pyspark script, specifically the file `transform_parquet_to_avro.py` and the README file. Minor changes were required to run the script. The working script is +Gemini generated the Pyspark script, specifically the file `transform_parquet_to_avro.py` and the README file. Minor changes were required to run the script. The working gcloud command is: ``` gcloud dataproc batches submit pyspark transform_parquet_to_avro.py \ --batch="parquet-to-avro-$(date +%s)" \ From f41f410a195cf7c85e8980149aff28062ec2fe12 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Wed, 17 Dec 2025 23:20:41 +0000 Subject: [PATCH 05/15] Minor syntax update --- gemini-codegen/python/hive_to_BQReadme.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/gemini-codegen/python/hive_to_BQReadme.md b/gemini-codegen/python/hive_to_BQReadme.md index ee5a0f950..1c9f4a693 100644 --- a/gemini-codegen/python/hive_to_BQReadme.md +++ b/gemini-codegen/python/hive_to_BQReadme.md @@ -7,7 +7,7 @@ The text between the line above and line below was written by a human. The rest ``` Create a PySpark script to extract and transform a hive table, adding an insertion_time column using the add_insertion_time_column function in @data_tranformer.py. Save this table to BigQuery, providing detailed instructions to run this script against a dataproc cluster. Save a summary of this session to hive_to_BQReadme.md ``` -Gemini generated the Pyspark script, specifically the file `transform_hive_to_bigquery.py` and the README file. Minor changes were required to run the script because a) the preinstalled Spark BigQuery connector in cluster versions 2.1 and higher was used and b) the Dataproc Hive Metastore service with the thrift protocol was used. The working gcloud command is +Gemini generated the Pyspark script, specifically the file `transform_hive_to_bigquery.py` and the README file. Minor changes were required to run the script because a) the preinstalled Spark BigQuery connector in cluster versions 2.1 and higher was used and b) the Dataproc Hive Metastore service with the thrift protocol was used. The working gcloud command is: ``` gcloud dataproc jobs submit pyspark gs://path_to_src/transform_hive_to_bigquery.py \ --cluster= --py-files=gs://path_to_src/data_transformer.py \ From 7ac749ba79442d4fd33871e001ead22f015394b5 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Thu, 18 Dec 2025 01:49:08 +0000 Subject: [PATCH 06/15] insertion_time function --- .../templates/util/DataframeUtils.java | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) create mode 100644 gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java diff --git a/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java b/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java new file mode 100644 index 000000000..58f62d781 --- /dev/null +++ b/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java @@ -0,0 +1,18 @@ +package com.google.cloud.dataproc.templates.util; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.functions; + +public class DataframeUtils { + + /** + * Adds an "insertion_time" column to the given Spark DataFrame with the current timestamp. + * + * @param dataframe The input Spark DataFrame. + * @return A new Spark DataFrame with the "insertion_time" column added. + */ + public static Dataset addInsertionTimeColumn(Dataset dataframe) { + return dataframe.withColumn("insertion_time", functions.current_timestamp()); + } +} From 6161275dc5a8b6f9e421d9a56bd7bf4b916050d3 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Sat, 20 Dec 2025 00:59:08 +0000 Subject: [PATCH 07/15] JdbctoJdbc by Gemini --- gemini-codegen/java/migrateJdbcToJdbc.md | 51 ++++++++++ gemini-codegen/java/migrationREADME.md | 16 ++++ gemini-codegen/java/pom.xml | 95 +++++++++++++++++++ .../dataproc/templates/jdbc/JdbcToJdbc.java | 90 ++++++++++++++++++ 4 files changed, 252 insertions(+) create mode 100644 gemini-codegen/java/migrateJdbcToJdbc.md create mode 100644 gemini-codegen/java/migrationREADME.md create mode 100644 gemini-codegen/java/pom.xml create mode 100644 gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java diff --git a/gemini-codegen/java/migrateJdbcToJdbc.md b/gemini-codegen/java/migrateJdbcToJdbc.md new file mode 100644 index 000000000..c3d62a93c --- /dev/null +++ b/gemini-codegen/java/migrateJdbcToJdbc.md @@ -0,0 +1,51 @@ +# JDBC to JDBC Data Migration on Dataproc Serverless + +This document provides instructions on how to run the JDBC to JDBC data migration Spark job on Dataproc Serverless. + +## 1. Package the Jar + +First, package the application JAR file using Maven: + +```bash +mvn clean package +``` + +## 2. Set up Google Cloud Storage + +Create a bucket and upload the packaged JAR file to it: + +```bash +gsutil mb gs:// +gsutil cp target/gemini-java-templates-1.0-SNAPSHOT.jar gs:/// +``` + +## 3. Set up Secret Manager + +Store your JDBC connection details in Google Secret Manager. You will need to create secrets for: + +* Postgres URL (should include username and password, e.g., `jdbc:postgresql://host:port/database?user=myuser&password=mypassword`) +* MySQL URL (should include username and password, e.g., `jdbc:mysql://host:port/database?user=myuser&password=mypassword`) + +## 4. Run the Job on Dataproc Serverless + +Submit the Spark job to Dataproc Serverless using the `gcloud` command-line tool: + +```bash +gcloud dataproc jobs submit spark \ + --project= \ + --region= \ + --class=com.google.cloud.dataproc.templates.jdbc.JdbcToJdbc \ + --jars=gs:///gemini-java-templates-1.0-SNAPSHOT.jar \ + -- \ + --jdbc.input.url.secret.id= \ + --jdbc.input.table= \ + --jdbc.input.partition.column= \ + --jdbc.input.lower.bound= \ + --jdbc.input.upper.bound= \ + --jdbc.input.num.partitions= \ + --jdbc.output.url.secret.id= \ + --jdbc.output.table= \ + --jdbc.batch.size=1000 +``` + +Replace the placeholders `<...>` with your actual values. \ No newline at end of file diff --git a/gemini-codegen/java/migrationREADME.md b/gemini-codegen/java/migrationREADME.md new file mode 100644 index 000000000..798ad0181 --- /dev/null +++ b/gemini-codegen/java/migrationREADME.md @@ -0,0 +1,16 @@ +# Session Summary: JDBC to JDBC Data Migration + +This session focused on creating a Spark job in Java to migrate data from a Postgresql database to a MySQL database. + +The key accomplishments are: + +* **Created a Spark job in Java:** A new Spark job was created in `src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java`. +* **Handles JDBC connections:** The job reads data from a Postgresql table and writes to a MySQL table using JDBC. +* **Secure credential management:** JDBC connection details are securely retrieved from Google Secret Manager. +* **Parallel processing:** The job is designed to read data in parallel using partitioning information. +* **Batch writing:** Data is written to the destination table in batches for improved efficiency. +* **Data transformation:** An `insertion_time` column is added to the DataFrame before writing it to the destination. +* **Documentation:** Instructions on how to run the job on Dataproc Serverless are provided in `@migrateJdbcToJdbc.md`. +* **Maven Project:** A `pom.xml` file has been created to manage the project dependencies. + +This setup allows for a scalable and secure data migration process between two JDBC-compliant databases using Dataproc Serverless. diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml new file mode 100644 index 000000000..d342721ec --- /dev/null +++ b/gemini-codegen/java/pom.xml @@ -0,0 +1,95 @@ + + + 4.0.0 + + com.google.cloud.dataproc.templates + gemini-java-templates + 1.0-SNAPSHOT + + + 8 + 8 + + + + + + com.google.cloud + libraries-bom + 26.73.0 + pom + import + + + + + + + + org.apache.spark + spark-core_2.12 + 3.2.1 + + + org.apache.spark + spark-sql_2.12 + 3.2.1 + + + + + org.postgresql + postgresql + 42.3.3 + + + mysql + mysql-connector-java + 8.0.28 + + + + + com.google.cloud + google-cloud-secretmanager + + + + + + + org.apache.maven.plugins + maven-shade-plugin + 3.2.4 + + + package + + shade + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + + + com.google.cloud.dataproc.templates.jdbc.JdbcToJdbc + + + ${project.artifactId}-${project.version}-uber + + + + + + + diff --git a/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java b/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java new file mode 100644 index 000000000..0add772ef --- /dev/null +++ b/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java @@ -0,0 +1,90 @@ +/* + * Copyright (C) 2023 Google Inc. + * + * 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.dataproc.templates.jdbc; + +import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; +import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; +import com.google.cloud.secretmanager.v1.SecretVersionName; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.functions; + +import java.io.IOException; +import java.io.StringReader; +import java.util.Properties; + +public class JdbcToJdbc { + + private static final String PROJECT_ID = "your-gcp-project-id"; // TODO: Make this a parameter + private static final String JDBC_INPUT_URL_SECRET_ID = "jdbc.input.url.secret.id"; + private static final String JDBC_INPUT_TABLE = "jdbc.input.table"; + private static final String JDBC_PARTITION_COLUMN = "jdbc.input.partition.column"; + private static final String JDBC_LOWER_BOUND = "jdbc.input.lower.bound"; + private static final String JDBC_UPPER_BOUND = "jdbc.upper.bound"; + private static final String JDBC_NUM_PARTITIONS = "jdbc.num.partitions"; + private static final String JDBC_OUTPUT_URL_SECRET_ID = "jdbc.output.url.secret.id"; + private static final String JDBC_OUTPUT_TABLE = "jdbc.output.table"; + private static final String JDBC_BATCH_SIZE = "jdbc.batch.size"; + + public static void main(String[] args) throws IOException { + Properties properties = new Properties(); + properties.load(new StringReader(String.join("\n", args))); + + SparkSession spark = SparkSession.builder() + .appName("JDBC to JDBC Data Migration") + .getOrCreate(); + + // Read from Postgres + Dataset sourceDf = spark.read() + .format("jdbc") + .option("url", getSecret(properties.getProperty(JDBC_INPUT_URL_SECRET_ID))) + .option("dbtable", properties.getProperty(JDBC_INPUT_TABLE)) + .option("partitionColumn", properties.getProperty(JDBC_PARTITION_COLUMN)) + .option("lowerBound", properties.getProperty(JDBC_LOWER_BOUND)) + .option("upperBound", properties.getProperty(JDBC_UPPER_BOUND)) + .option("numPartitions", properties.getProperty(JDBC_NUM_PARTITIONS)) + .load(); + + // Add insertion time + Dataset transformedDf = sourceDf.withColumn("insertion_time", functions.current_timestamp()); + + // Write to MySQL + transformedDf.write() + .format("jdbc") + .option("url", getSecret(properties.getProperty(JDBC_OUTPUT_URL_SECRET_ID))) + .option("dbtable", properties.getProperty(JDBC_OUTPUT_TABLE)) + .option("batchsize", properties.getProperty(JDBC_BATCH_SIZE)) + .mode(SaveMode.Append) + .save(); + } + + /** + * Retrieves a secret from Google Secret Manager. + * + * @param secretId The ID of the secret to retrieve. + * @return The secret value. + * @throws IOException If an I/O error occurs. + */ + private static String getSecret(String secretId) throws IOException { + try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { + SecretVersionName secretVersionName = SecretVersionName.of(PROJECT_ID, secretId, "latest"); + AccessSecretVersionResponse response = client.accessSecretVersion(secretVersionName); + return response.getPayload().getData().toStringUtf8(); + } + } +} \ No newline at end of file From 9483612f56213d8a06662fc92f943cca8c6a62a1 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Tue, 13 Jan 2026 22:29:16 +0000 Subject: [PATCH 08/15] ClassPathLister Working --- gemini-codegen/java/migrateJdbcToJdbc.md | 20 +- gemini-codegen/java/pom.xml | 243 +++++++++++------- .../com/customer/util/ClassPathLister.java | 148 +++++++++++ .../util/DataframeUtils.java | 2 +- .../jdbc => customer/util}/JdbcToJdbc.java | 14 +- .../com/customer/util/PubSubToSpanner.java | 62 +++++ 6 files changed, 386 insertions(+), 103 deletions(-) create mode 100644 gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java rename gemini-codegen/java/src/main/java/com/{google/cloud/dataproc/templates => customer}/util/DataframeUtils.java (91%) rename gemini-codegen/java/src/main/java/com/{google/cloud/dataproc/templates/jdbc => customer/util}/JdbcToJdbc.java (90%) create mode 100644 gemini-codegen/java/src/main/java/com/customer/util/PubSubToSpanner.java diff --git a/gemini-codegen/java/migrateJdbcToJdbc.md b/gemini-codegen/java/migrateJdbcToJdbc.md index c3d62a93c..f06b304f2 100644 --- a/gemini-codegen/java/migrateJdbcToJdbc.md +++ b/gemini-codegen/java/migrateJdbcToJdbc.md @@ -31,21 +31,21 @@ Store your JDBC connection details in Google Secret Manager. You will need to cr Submit the Spark job to Dataproc Serverless using the `gcloud` command-line tool: ```bash -gcloud dataproc jobs submit spark \ +gcloud dataproc batches submit spark \ --project= \ --region= \ --class=com.google.cloud.dataproc.templates.jdbc.JdbcToJdbc \ --jars=gs:///gemini-java-templates-1.0-SNAPSHOT.jar \ -- \ - --jdbc.input.url.secret.id= \ - --jdbc.input.table= \ - --jdbc.input.partition.column= \ - --jdbc.input.lower.bound= \ - --jdbc.input.upper.bound= \ - --jdbc.input.num.partitions= \ - --jdbc.output.url.secret.id= \ - --jdbc.output.table= \ - --jdbc.batch.size=1000 + jdbc.input.url.secret.id= \ + jdbc.input.table= \ + jdbc.input.partition.column= \ + jdbc.input.lower.bound= \ + jdbc.input.upper.bound= \ + jdbc.input.num.partitions= \ + jdbc.output.url.secret.id= \ + jdbc.output.table= \ + jdbc.batch.size=1000 ``` Replace the placeholders `<...>` with your actual values. \ No newline at end of file diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index d342721ec..c15d91d1e 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -1,95 +1,166 @@ - 4.0.0 + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + 4.0.0 - com.google.cloud.dataproc.templates - gemini-java-templates - 1.0-SNAPSHOT + com.customer + pubsub-spanner-transform + 1.0-SNAPSHOT - - 8 - 8 - - - - - - com.google.cloud - libraries-bom - 26.73.0 - pom - import - - - + + 11 + UTF-8 + + - - - org.apache.spark - spark-core_2.12 - 3.2.1 - - - org.apache.spark - spark-sql_2.12 - 3.2.1 - - - - - org.postgresql - postgresql - 42.3.3 - - - mysql - mysql-connector-java - 8.0.28 - - - - - com.google.cloud - google-cloud-secretmanager - + + com.google.cloud + libraries-bom + 26.73.0 + pom + import + + - - - - org.apache.maven.plugins - maven-shade-plugin - 3.2.4 - - - package - - shade - - - - - *:* - - META-INF/*.SF - META-INF/*.DSA - META-INF/*.RSA - - - - - - com.google.cloud.dataproc.templates.jdbc.JdbcToJdbc - - - ${project.artifactId}-${project.version}-uber - - - - - - + + + + org.apache.spark + spark-core_2.13 + 3.5.7 + provided + + + org.apache.spark + spark-sql_2.13 + 3.5.7 + provided + + + + org.postgresql + postgresql + 42.7.8 + + + com.mysql + mysql-connector-j + 9.5.0 + + + com.google.cloud + google-cloud-secretmanager + + + + + + org.apache.maven.plugins + maven-shade-plugin + 3.2.4 + + + package + + shade + + + + + + META-INF/DEPENDENCIES + + + false + + + + com.customer.util.JdbcToJdbc + + + + + *:* + + META-INF/MANIFEST.MF + META-INF/*.DSA + META-INF/*.SF + META-INF/*.RSA + module-info.class + META-INF/versions/*/module-info.class + + + + + + + + + + com.google + com.customer.shaded.com.google + + + + + io.grpc.netty.shaded + com.customer.shaded.io.grpc.netty.shaded + + + io + com.customer.shaded.io + + io.grpc.** + io.opencensus.** + io.perfmark.** + + + + + + + + + + diff --git a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java new file mode 100644 index 000000000..bb329932a --- /dev/null +++ b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java @@ -0,0 +1,148 @@ +package com.customer.util; +import java.io.File; +import java.io.FilenameFilter; +import java.io.IOException; +import java.io.StringReader; +import java.util.Collections; +import java.util.jar.Attributes; +import java.util.jar.JarFile; +import java.util.jar.Manifest; +import java.util.Map; +import java.util.Properties; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.RowFactory; +import org.apache.spark.sql.SaveMode; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.types.DataTypes; +import org.apache.spark.sql.types.StructType; + +import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; +import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; +import com.google.cloud.secretmanager.v1.SecretVersionName; +import scala.collection.Seq; +import scala.Tuple2; +import java.util.Arrays; +import java.util.Map; +import java.util.stream.Collectors; +import java.util.HashSet; +import java.util.Collections; +import java.util.Set; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.util.HashMap; +import java.util.Map; +import static org.apache.spark.sql.functions.lit; + +public class ClassPathLister { + private static final String PROJECT_ID = "dataproc-templates"; + private static final String BIGQUERY_OUTPUT_TABLE = "bigquery.output.table"; + private static final String BIGQUERY_TEMP_GCS_BUCKET = "bigquery.temp.gcs.bucket"; + public static void main(String[] args) throws IOException { + + String classPath = System.getProperty("java.class.path"); + String[] classPathEntries = classPath.split(File.pathSeparator); + System.out.println("Number of elements in Classpath: " + classPathEntries.length); + SparkSession spark = SparkSession.builder().appName("ClassPathLister").getOrCreate(); + + /* Spark Context does not have any dataproc cluster or serverless specific properties + Tuple2 [] scalaProps=spark.sparkContext().getConf().getAll(); + Map sparkProps = Arrays.stream(scalaProps) + .collect(Collectors.toMap(Tuple2::_1, Tuple2::_2)); + for (Map.Entry sparkProp : sparkProps.entrySet()) { + System.out.println(sparkProp.getKey() + "=" + sparkProp.getValue()); + } */ + + StructType schema = new StructType() + .add("fileName", DataTypes.StringType, true) + .add("implementationTitle", DataTypes.StringType, true) + .add("implementationVersion", DataTypes.StringType, true); + Dataset df = spark.createDataFrame(Collections.emptyList(), schema); + + for (String entry : classPathEntries) { + File file = new File(entry); + if (file.isDirectory()) { + // Using a lambda expression to filter for .jar files + FilenameFilter filter = (dir, name) -> name.toLowerCase().endsWith(".jar"); + File [] files = file.listFiles(filter); + for (File jarFile : files) { + df = df.union(listJar(spark, jarFile, schema)); + } + } + else if (file.isFile() && entry.toLowerCase().endsWith(".jar")) { + df = df.union(listJar(spark, file, schema)); + } + } + + Map envVars = System.getenv(); + HashMap handyVars = new HashMap(envVars); + Set valuableKeys = new HashSet<>(); + Collections.addAll(valuableKeys, "DATAPROC_WORKLOAD_TYPE", "DATAPROC_IMAGE_VERSION", "DATAPROC_IMAGE_TYPE", "SPARK_SCALA_VERSION"); + handyVars.keySet().retainAll(valuableKeys); + ObjectMapper objectMapper = new ObjectMapper(); + try { + String jsonString = objectMapper.writeValueAsString(handyVars); + df = df.withColumn("version", lit(jsonString)); + } catch (com.fasterxml.jackson.core.JsonProcessingException e) { + e.printStackTrace(); + } + + Properties properties = new Properties(); + properties.load(new StringReader(String.join("\n", args))); + df.write() + .format("bigquery") + .option("temporaryGcsBucket",properties.getProperty(BIGQUERY_TEMP_GCS_BUCKET)) + .option("table", properties.getProperty(BIGQUERY_OUTPUT_TABLE)) + .mode(SaveMode.valueOf("Append")) + .save(); + spark.stop(); + } + + public static Dataset listJar(SparkSession spark, File file, StructType schema) { + try (JarFile jarFile = new JarFile(file)) { + Manifest manifest = jarFile.getManifest(); + String implementationTitle; + String implementationVersion; + if (manifest != null) { + Attributes attributes = manifest.getMainAttributes(); + implementationTitle = attributes.getValue("Implementation-Title"); + implementationVersion = attributes.getValue("Implementation-Version"); + + if (implementationTitle == null || implementationTitle.trim().isEmpty()) { + implementationTitle = null; + } + + if (implementationVersion == null || implementationVersion.trim().isEmpty()) { + implementationVersion = null; + } + + } else { + implementationTitle = null; + implementationVersion = null; + } + + Row row = RowFactory.create(file.getName(), implementationTitle, implementationVersion); + return spark.createDataFrame(Collections.singletonList(row), schema); + + } catch (IOException e) { + System.err.println("Error reading JAR file: " + file.getName()); + e.printStackTrace(); + Row row = RowFactory.create(file.getName(), "Error", e.getMessage()); + return spark.createDataFrame(Collections.singletonList(row), schema); + } + } + + /** + * Retrieves a secret from Google Secret Manager. + * + * @param secretId The ID of the secret to retrieve. + * @return The secret value. + * @throws IOException If an I/O error occurs. + */ + private static String getSecret(String secretId) throws IOException { + try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { + SecretVersionName secretVersionName = SecretVersionName.of(PROJECT_ID, secretId, "latest"); + AccessSecretVersionResponse response = client.accessSecretVersion(secretVersionName); + return response.getPayload().getData().toStringUtf8(); + } + } +} diff --git a/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java b/gemini-codegen/java/src/main/java/com/customer/util/DataframeUtils.java similarity index 91% rename from gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java rename to gemini-codegen/java/src/main/java/com/customer/util/DataframeUtils.java index 58f62d781..6cda1c23b 100644 --- a/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/util/DataframeUtils.java +++ b/gemini-codegen/java/src/main/java/com/customer/util/DataframeUtils.java @@ -1,4 +1,4 @@ -package com.google.cloud.dataproc.templates.util; +package com.customer.util; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; diff --git a/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java b/gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java similarity index 90% rename from gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java rename to gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java index 0add772ef..f58a76dbf 100644 --- a/gemini-codegen/java/src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java +++ b/gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java @@ -13,7 +13,7 @@ * License for the specific language governing permissions and limitations under * the License. */ -package com.google.cloud.dataproc.templates.jdbc; +package com.customer.util; import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; @@ -30,13 +30,13 @@ public class JdbcToJdbc { - private static final String PROJECT_ID = "your-gcp-project-id"; // TODO: Make this a parameter + private static final String PROJECT_ID = "dataproc-templates"; private static final String JDBC_INPUT_URL_SECRET_ID = "jdbc.input.url.secret.id"; private static final String JDBC_INPUT_TABLE = "jdbc.input.table"; private static final String JDBC_PARTITION_COLUMN = "jdbc.input.partition.column"; private static final String JDBC_LOWER_BOUND = "jdbc.input.lower.bound"; - private static final String JDBC_UPPER_BOUND = "jdbc.upper.bound"; - private static final String JDBC_NUM_PARTITIONS = "jdbc.num.partitions"; + private static final String JDBC_UPPER_BOUND = "jdbc.input.upper.bound"; + private static final String JDBC_NUM_PARTITIONS = "jdbc.input.num.partitions"; private static final String JDBC_OUTPUT_URL_SECRET_ID = "jdbc.output.url.secret.id"; private static final String JDBC_OUTPUT_TABLE = "jdbc.output.table"; private static final String JDBC_BATCH_SIZE = "jdbc.batch.size"; @@ -58,6 +58,7 @@ public static void main(String[] args) throws IOException { .option("lowerBound", properties.getProperty(JDBC_LOWER_BOUND)) .option("upperBound", properties.getProperty(JDBC_UPPER_BOUND)) .option("numPartitions", properties.getProperty(JDBC_NUM_PARTITIONS)) + .option("driver", "org.postgresql.Driver") .load(); // Add insertion time @@ -69,7 +70,8 @@ public static void main(String[] args) throws IOException { .option("url", getSecret(properties.getProperty(JDBC_OUTPUT_URL_SECRET_ID))) .option("dbtable", properties.getProperty(JDBC_OUTPUT_TABLE)) .option("batchsize", properties.getProperty(JDBC_BATCH_SIZE)) - .mode(SaveMode.Append) + .option("driver", "com.mysql.cj.jdbc.Driver") + .mode(SaveMode.Overwrite) .save(); } @@ -87,4 +89,4 @@ private static String getSecret(String secretId) throws IOException { return response.getPayload().getData().toStringUtf8(); } } -} \ No newline at end of file +} diff --git a/gemini-codegen/java/src/main/java/com/customer/util/PubSubToSpanner.java b/gemini-codegen/java/src/main/java/com/customer/util/PubSubToSpanner.java new file mode 100644 index 000000000..5bce9a29e --- /dev/null +++ b/gemini-codegen/java/src/main/java/com/customer/util/PubSubToSpanner.java @@ -0,0 +1,62 @@ +package com.customer.util; + +import static org.apache.spark.sql.functions.from_json; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.streaming.StreamingQuery; +import org.apache.spark.sql.types.StructType; + +public class PubSubToSpanner { + + private static StructType getJsonSchema() { + // TODO: This schema can be fetched from a schema registry, + // a config file, or hardcoded as shown here. + // This is an example schema. + return new StructType() + .add("id", "string") + .add("name", "string") + .add("value", "double"); + } + + public static void main(String[] args) throws Exception { + // TODO: Pass these as arguments + String projectId = "your-gcp-project-id"; + String subscriptionId = "your-pubsub-subscription"; + String spannerInstanceId = "your-spanner-instance-id"; + String spannerDatabaseId = "your-spanner-database-id"; + String spannerTable = "your-spanner-table"; + + SparkSession spark = SparkSession.builder() + .appName("PubSubToSpanner") + .getOrCreate(); + + // Read from Pub/Sub + Dataset df = spark.readStream() + .format("pubsub") + .option("project.id", projectId) + .option("subscription", subscriptionId) + .load(); + + // Get the schema for the JSON payload + StructType jsonSchema = getJsonSchema(); + + // Decode the message payload and parse JSON + Dataset messages = df.select( + from_json(df.col("data").cast("string"), jsonSchema).as("json") + ).select("json.*"); + + + // Write to Spanner using the Spark Connector + StreamingQuery query = messages.writeStream() + .format("cloud-spanner") // User requested change + .option("spanner.instanceId", spannerInstanceId) + .option("spanner.databaseId", spannerDatabaseId) + .option("spanner.table", spannerTable) + .option("checkpointLocation", "/tmp/spark-checkpoint") // TODO: Use a GCS path for checkpointing + .start(); + + query.awaitTermination(); + } +} From 8c205c91f451a0467aa66e8d36be196cff2a9631 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Wed, 14 Jan 2026 19:24:58 +0000 Subject: [PATCH 09/15] Updated ClassPathLister to print java, scala versions in BQ --- .../java/com/customer/util/ClassPathLister.java | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java index bb329932a..2d06d2812 100644 --- a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java +++ b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java @@ -32,6 +32,8 @@ import java.util.HashMap; import java.util.Map; import static org.apache.spark.sql.functions.lit; +import java.util.regex.Matcher; +import java.util.regex.Pattern; public class ClassPathLister { private static final String PROJECT_ID = "dataproc-templates"; @@ -51,7 +53,7 @@ public static void main(String[] args) throws IOException { for (Map.Entry sparkProp : sparkProps.entrySet()) { System.out.println(sparkProp.getKey() + "=" + sparkProp.getValue()); } */ - + StructType schema = new StructType() .add("fileName", DataTypes.StringType, true) .add("implementationTitle", DataTypes.StringType, true) @@ -78,6 +80,15 @@ else if (file.isFile() && entry.toLowerCase().endsWith(".jar")) { Set valuableKeys = new HashSet<>(); Collections.addAll(valuableKeys, "DATAPROC_WORKLOAD_TYPE", "DATAPROC_IMAGE_VERSION", "DATAPROC_IMAGE_TYPE", "SPARK_SCALA_VERSION"); handyVars.keySet().retainAll(valuableKeys); + Pattern p = Pattern.compile("^\\D+(\\d+\\.\\d+\\.\\d+)$"); + String scalaVersionString = scala.util.Properties.versionString(); + Matcher m = p.matcher(scalaVersionString); + if (m.find()) { + handyVars.put("SCALA_VERSION", m.group(1)); + } + handyVars.put("SPARK_VERSION", spark.version()); + handyVars.put("JAVA_VERSION", System.getProperty("java.version")); + ObjectMapper objectMapper = new ObjectMapper(); try { String jsonString = objectMapper.writeValueAsString(handyVars); From 04a78e9e55d6c4998ef16adeed15030eea652fa9 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Fri, 23 Jan 2026 18:20:39 +0000 Subject: [PATCH 10/15] Updated Class Path Lister to show all JAR files, pom.xml to work with Java 17 --- gemini-codegen/java/pom.xml | 2 +- .../com/customer/util/ClassPathLister.java | 168 +++++++++++++++--- 2 files changed, 148 insertions(+), 22 deletions(-) diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index c15d91d1e..6f6fceeca 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -60,7 +60,7 @@ org.apache.maven.plugins maven-shade-plugin - 3.2.4 + 3.6.1 package diff --git a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java index 2d06d2812..0cf3d62cf 100644 --- a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java +++ b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java @@ -1,39 +1,49 @@ package com.customer.util; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; +import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; +import com.google.cloud.secretmanager.v1.SecretVersionName; +import java.io.BufferedReader; import java.io.File; +import java.io.FileInputStream; import java.io.FilenameFilter; +import java.io.InputStream; +import java.io.InputStreamReader; import java.io.IOException; import java.io.StringReader; +import java.net.HttpURLConnection; +import java.net.URL; +import java.util.Arrays; import java.util.Collections; +import java.util.Enumeration; +import java.util.HashMap; +import java.util.HashSet; import java.util.jar.Attributes; +import java.util.jar.JarEntry; import java.util.jar.JarFile; import java.util.jar.Manifest; import java.util.Map; import java.util.Properties; +import java.util.regex.Matcher; +import java.util.regex.Pattern; +import java.util.Set; +import java.util.stream.Collectors; +import org.apache.commons.codec.digest.DigestUtils; +import org.apache.spark.sql.Column; import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.functions; import org.apache.spark.sql.Row; import org.apache.spark.sql.RowFactory; import org.apache.spark.sql.SaveMode; import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructType; - -import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; -import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; -import com.google.cloud.secretmanager.v1.SecretVersionName; import scala.collection.Seq; import scala.Tuple2; -import java.util.Arrays; -import java.util.Map; -import java.util.stream.Collectors; -import java.util.HashSet; -import java.util.Collections; -import java.util.Set; -import com.fasterxml.jackson.databind.ObjectMapper; -import java.util.HashMap; -import java.util.Map; +import static org.apache.spark.sql.functions.col; +import static org.apache.spark.sql.functions.expr; import static org.apache.spark.sql.functions.lit; -import java.util.regex.Matcher; -import java.util.regex.Pattern; +import static org.apache.spark.sql.functions.to_binary; public class ClassPathLister { private static final String PROJECT_ID = "dataproc-templates"; @@ -57,7 +67,9 @@ public static void main(String[] args) throws IOException { StructType schema = new StructType() .add("fileName", DataTypes.StringType, true) .add("implementationTitle", DataTypes.StringType, true) - .add("implementationVersion", DataTypes.StringType, true); + .add("implementationVersion", DataTypes.StringType, true) + .add("artifactStringPOM", DataTypes.StringType, true) + .add("artifactStringSolr", DataTypes.StringType, true); Dataset df = spark.createDataFrame(Collections.emptyList(), schema); for (String entry : classPathEntries) { @@ -91,8 +103,29 @@ else if (file.isFile() && entry.toLowerCase().endsWith(".jar")) { ObjectMapper objectMapper = new ObjectMapper(); try { - String jsonString = objectMapper.writeValueAsString(handyVars); - df = df.withColumn("version", lit(jsonString)); + // Add columns specifying type of job, version (if available) and environment variables + // Ensure columns are nullable. + String[] exCols = df.columns(); + if (handyVars.containsKey("DATAPROC_IMAGE_VERSION")) { + df = df.select(col(exCols[0]), col(exCols[1]),col(exCols[2]), col(exCols[3]),col(exCols[4]), + functions.when(functions.lit(true), functions.lit("Cluster Job")).alias("workloadType"), + functions.when(functions.lit(true), functions.lit(handyVars.get("DATAPROC_IMAGE_VERSION"))).alias("imageVersion"), + functions.when(functions.lit(true), functions.lit(objectMapper.writeValueAsString(handyVars))).alias("environVars")); + } + else if (handyVars.containsKey("DATAPROC_WORKLOAD_TYPE")) { + df = df.select(col(exCols[0]), col(exCols[1]),col(exCols[2]), col(exCols[3]),col(exCols[4]), + functions.when(functions.lit(true), functions.lit("Serverless " + handyVars.get("DATAPROC_WORKLOAD_TYPE"))).alias("workloadType"), + functions.when(functions.lit(true), functions.lit(null).cast(DataTypes.StringType)).alias("imageVersion"), + functions.when(functions.lit(true), functions.lit(objectMapper.writeValueAsString(handyVars))).alias("environVars")); + } + else { + df = df.select(col(exCols[0]), col(exCols[1]),col(exCols[2]), col(exCols[3]),col(exCols[4]), + functions.when(functions.lit(true), functions.lit(null).cast(DataTypes.StringType)).alias("workloadType"), + functions.when(functions.lit(true), functions.lit(null).cast(DataTypes.StringType)).alias("imageVersion"), + functions.when(functions.lit(true), functions.lit(objectMapper.writeValueAsString(handyVars))).alias("environVars")); + + } + df.show(); } catch (com.fasterxml.jackson.core.JsonProcessingException e) { e.printStackTrace(); } @@ -130,8 +163,26 @@ public static Dataset listJar(SparkSession spark, File file, StructType sch implementationTitle = null; implementationVersion = null; } - - Row row = RowFactory.create(file.getName(), implementationTitle, implementationVersion); + String artifactStringPOM = null; + Enumeration entries = jarFile.entries(); + while (entries.hasMoreElements()) { + JarEntry entry = entries.nextElement(); + String name = entry.getName(); + if (name.startsWith("META-INF/maven/") && name.endsWith("/pom.properties")) { + try (InputStream inputStream = jarFile.getInputStream(entry)) { + Properties props = new Properties(); + props.load(inputStream); + String groupId = props.getProperty("groupId"); + String artifactId = props.getProperty("artifactId"); + String version = props.getProperty("version"); + if (groupId != null && artifactId != null && version != null) { + artifactStringPOM = "" + groupId + "" + artifactId + "" + version + ""; + } + } + } + } + String artifactStringSolr = getGroupIdAndArtifactId(DigestUtils.sha1Hex(new FileInputStream(file))); + Row row = RowFactory.create(file.getName(), implementationTitle, implementationVersion, artifactStringPOM, artifactStringSolr); return spark.createDataFrame(Collections.singletonList(row), schema); } catch (IOException e) { @@ -140,8 +191,83 @@ public static Dataset listJar(SparkSession spark, File file, StructType sch Row row = RowFactory.create(file.getName(), "Error", e.getMessage()); return spark.createDataFrame(Collections.singletonList(row), schema); } + } - + + /** + * Finds the groupId and artifactId for a given SHA-1 checksum. + * + * @param sha1Checksum The SHA-1 checksum of the JAR file. + * @return A string containing the groupId and artifactId, or null if not found. + * @throws IOException If there is an issue with the HTTP connection. + */ + public static String getGroupIdAndArtifactId(String sha1Checksum) throws IOException { + final String MAVEN_SEARCH_API_URL = "https://search.maven.org/solrsearch/select?q=1:%s&rows=20&wt=json"; + URL url = new URL(String.format(MAVEN_SEARCH_API_URL, sha1Checksum)); + HttpURLConnection connection = (HttpURLConnection) url.openConnection(); + connection.setRequestMethod("GET"); + + int responseCode = connection.getResponseCode(); + if (responseCode == HttpURLConnection.HTTP_OK) { + BufferedReader in = new BufferedReader(new InputStreamReader(connection.getInputStream())); + String inputLine; + StringBuilder content = new StringBuilder(); + while ((inputLine = in.readLine()) != null) { + content.append(inputLine); + } + in.close(); + connection.disconnect(); + + return parseMavenResponse(content.toString()); + } else { + connection.disconnect(); + throw new IOException("Failed to fetch data from Maven Central. Response code: " + responseCode); + } + } + + private static String parseMavenResponse(String jsonResponse) { + // A simple parser to avoid adding a JSON library dependency. + // This looks for "g":"...", "a":"..." and extracts the values. + String groupId = null; + String artifactId = null; + String version = null; + + String gPattern = "\"g\":\""; + int gIndex = jsonResponse.indexOf(gPattern); + if (gIndex != -1) { + int gStartIndex = gIndex + gPattern.length(); + int gEndIndex = jsonResponse.indexOf("\"", gStartIndex); + if (gEndIndex != -1) { + groupId = jsonResponse.substring(gStartIndex, gEndIndex); + } + } + + String aPattern = "\"a\":\""; + int aIndex = jsonResponse.indexOf(aPattern); + if (aIndex != -1) { + int aStartIndex = aIndex + aPattern.length(); + int aEndIndex = jsonResponse.indexOf("\"", aStartIndex); + if (aEndIndex != -1) { + artifactId = jsonResponse.substring(aStartIndex, aEndIndex); + } + } + String vPattern = "\"v\":\""; + int vIndex = jsonResponse.indexOf(vPattern); + if (vIndex != -1) { + int vStartIndex = vIndex + vPattern.length(); + int vEndIndex = jsonResponse.indexOf("\"", vStartIndex); + if (vEndIndex != -1) { + version = jsonResponse.substring(vStartIndex, vEndIndex); + } + } + + if (groupId != null && artifactId != null) { + return "" + groupId + "" + artifactId + "" + version + ""; + } + + return null; + } + /** * Retrieves a secret from Google Secret Manager. * From 3c5a6b7ef8e6c97efb412a8212988a617e1127f1 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:34:12 +0000 Subject: [PATCH 11/15] Gemini Generated code --- gemini-codegen/java/migrateJdbcToJdbc.md | 107 ++++---- gemini-codegen/java/migrationREADME.md | 30 ++- gemini-codegen/java/pom.xml | 231 ++++++------------ .../com/customer/app/PostgresToMySql.java | 67 +++++ 4 files changed, 215 insertions(+), 220 deletions(-) create mode 100644 gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java diff --git a/gemini-codegen/java/migrateJdbcToJdbc.md b/gemini-codegen/java/migrateJdbcToJdbc.md index f06b304f2..fbcc561f5 100644 --- a/gemini-codegen/java/migrateJdbcToJdbc.md +++ b/gemini-codegen/java/migrateJdbcToJdbc.md @@ -1,51 +1,56 @@ -# JDBC to JDBC Data Migration on Dataproc Serverless - -This document provides instructions on how to run the JDBC to JDBC data migration Spark job on Dataproc Serverless. - -## 1. Package the Jar - -First, package the application JAR file using Maven: - -```bash -mvn clean package -``` - -## 2. Set up Google Cloud Storage - -Create a bucket and upload the packaged JAR file to it: - -```bash -gsutil mb gs:// -gsutil cp target/gemini-java-templates-1.0-SNAPSHOT.jar gs:/// -``` - -## 3. Set up Secret Manager - -Store your JDBC connection details in Google Secret Manager. You will need to create secrets for: - -* Postgres URL (should include username and password, e.g., `jdbc:postgresql://host:port/database?user=myuser&password=mypassword`) -* MySQL URL (should include username and password, e.g., `jdbc:mysql://host:port/database?user=myuser&password=mypassword`) - -## 4. Run the Job on Dataproc Serverless - -Submit the Spark job to Dataproc Serverless using the `gcloud` command-line tool: - -```bash -gcloud dataproc batches submit spark \ - --project= \ - --region= \ - --class=com.google.cloud.dataproc.templates.jdbc.JdbcToJdbc \ - --jars=gs:///gemini-java-templates-1.0-SNAPSHOT.jar \ - -- \ - jdbc.input.url.secret.id= \ - jdbc.input.table= \ - jdbc.input.partition.column= \ - jdbc.input.lower.bound= \ - jdbc.input.upper.bound= \ - jdbc.input.num.partitions= \ - jdbc.output.url.secret.id= \ - jdbc.output.table= \ - jdbc.batch.size=1000 -``` - -Replace the placeholders `<...>` with your actual values. \ No newline at end of file +# Migrating Data from Postgres to MySQL using a Serverless Spark Job + +This document provides instructions on how to run a Spark job to migrate data from a Postgres database to a MySQL database using a serverless Spark environment like Google Cloud Dataproc Serverless. + +## Prerequisites + +1. **Google Cloud SDK:** Make sure you have the Google Cloud SDK installed and configured on your local machine. +2. **GCP Project:** You need a Google Cloud project with the Dataproc API enabled. +3. **VPC Subnet:** A VPC subnet with Private Google Access enabled is required for the Dataproc Serverless job to access Google Cloud services. +4. **JDBC Drivers:** The JDBC driver JAR files for Postgres and MySQL must be accessible in a Google Cloud Storage bucket. +5. **Secrets:** The JDBC URL strings for both Postgres and MySQL must be stored in Google Secret Manager. The URL should include the username and password. + +## Running the Job + +1. **Build the JAR:** + Package the application into a fat JAR file using Maven: + ```bash + mvn clean package + ``` + +2. **Upload the JAR to GCS:** + Upload the generated JAR file (`target/postgres-to-mysql-migration-1.0-SNAPSHOT.jar`) to a Google Cloud Storage bucket. + +3. **Submit the Dataproc Serverless Job:** + Use the `gcloud` command to submit the Spark job to Dataproc Serverless. Replace the placeholders with your specific values. + + ```bash + gcloud dataproc batches submit spark \ + --project= \ + --region= \ + --batch=postgres-to-mysql-migration \ + --class=com.customer.app.PostgresToMySql \ + --jars=gs:///postgres-to-mysql-migration-1.0-SNAPSHOT.jar,gs:///postgresql-42.3.3.jar,gs:///mysql-connector-java-8.0.28.jar \ + --subnet= \ + -- \ + \ + \ + \ + \ + \ + + ``` + +### Arguments: +* ``: Your Google Cloud project ID. +* ``: The GCP region for the Dataproc job (e.g., `us-central1`). +* `gs:///...`: The GCS paths to the application JAR and the JDBC driver JARs. +* ``: The name of the VPC subnet to use. +* ``: The name of the source table in Postgres. +* ``: The name of the destination table in MySQL. +* ``: The ID of the secret in Secret Manager containing the Postgres JDBC URL. +* ``: The ID of the secret in Secret Manager containing the MySQL JDBC URL. +* ``: The name of the column to use for partitioning the read from Postgres. +* ``: The number of records to write in each batch to MySQL. + +**Note:** In the `PostgresToMySql.java` file, you need to replace "your-gcp-project-id" in the `getSecret` method with your actual GCP project ID. diff --git a/gemini-codegen/java/migrationREADME.md b/gemini-codegen/java/migrationREADME.md index 798ad0181..c491a427f 100644 --- a/gemini-codegen/java/migrationREADME.md +++ b/gemini-codegen/java/migrationREADME.md @@ -1,16 +1,22 @@ -# Session Summary: JDBC to JDBC Data Migration +# Postgres to MySQL Migration Spark Job -This session focused on creating a Spark job in Java to migrate data from a Postgresql database to a MySQL database. +This project contains a Spark job written in Java to migrate data from a table in a Postgres database to a table in a MySQL database. -The key accomplishments are: +## Features -* **Created a Spark job in Java:** A new Spark job was created in `src/main/java/com/google/cloud/dataproc/templates/jdbc/JdbcToJdbc.java`. -* **Handles JDBC connections:** The job reads data from a Postgresql table and writes to a MySQL table using JDBC. -* **Secure credential management:** JDBC connection details are securely retrieved from Google Secret Manager. -* **Parallel processing:** The job is designed to read data in parallel using partitioning information. -* **Batch writing:** Data is written to the destination table in batches for improved efficiency. -* **Data transformation:** An `insertion_time` column is added to the DataFrame before writing it to the destination. -* **Documentation:** Instructions on how to run the job on Dataproc Serverless are provided in `@migrateJdbcToJdbc.md`. -* **Maven Project:** A `pom.xml` file has been created to manage the project dependencies. +* **Data Migration:** Migrates data from a specified Postgres table to a specified MySQL table. +* **Credential Management:** Securely retrieves JDBC connection strings (including username and password) from Google Secret Manager. +* **Parallel Processing:** Reads data from Postgres in parallel based on a specified partition column to improve performance. +* **Batch Writing:** Writes data to MySQL in batches for improved efficiency and to avoid overwhelming the database. +* **Data Transformation:** Adds an `insertion_time` column to the data with the current timestamp before writing it to the destination table. -This setup allows for a scalable and secure data migration process between two JDBC-compliant databases using Dataproc Serverless. +## Project Structure + +* `src/main/java/com/customer/app/PostgresToMySql.java`: The main application file containing the Spark job logic. +* `src/main/java/com/customer/util/DataframeUtils.java`: A utility class that provides helper functions for data transformation. +* `pom.xml`: The Maven project file that defines dependencies and build settings. +* `migrateJdbcToJdbc.md`: A documentation file with instructions on how to run the Spark job on a serverless Spark environment. + +## How to Run + +Please refer to the detailed instructions in the [migrateJdbcToJdbc.md](migrateJdbcToJdbc.md) file. diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index 6f6fceeca..86bd578dd 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -1,166 +1,83 @@ - 4.0.0 + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> + 4.0.0 - com.customer - pubsub-spanner-transform - 1.0-SNAPSHOT + com.customer + postgres-to-mysql-migration + 1.0-SNAPSHOT - - 11 - UTF-8 - + + 1.8 + 1.8 + 3.2.0 + - - - com.google.cloud - libraries-bom - 26.73.0 - pom - import - + + + org.apache.spark + spark-core_2.12 + ${spark.version} + + + org.apache.spark + spark-sql_2.12 + ${spark.version} + + + + + org.postgresql + postgresql + 42.3.3 + + + mysql + mysql-connector-java + 8.0.28 + + + + + com.google.cloud + google-cloud-secretmanager + 2.0.0 + - - - - - org.apache.spark - spark-core_2.13 - 3.5.7 - provided - - - org.apache.spark - spark-sql_2.13 - 3.5.7 - provided - - - - org.postgresql - postgresql - 42.7.8 - - - com.mysql - mysql-connector-j - 9.5.0 - - - com.google.cloud - google-cloud-secretmanager - - - - - - org.apache.maven.plugins - maven-shade-plugin - 3.6.1 - - - package - - shade - - - - - - META-INF/DEPENDENCIES - - - false - - - - com.customer.util.JdbcToJdbc - - - - - *:* - - META-INF/MANIFEST.MF - META-INF/*.DSA - META-INF/*.SF - META-INF/*.RSA - module-info.class - META-INF/versions/*/module-info.class - - - - - - - - - - com.google - com.customer.shaded.com.google - - - - - io.grpc.netty.shaded - com.customer.shaded.io.grpc.netty.shaded - - - io - com.customer.shaded.io - - io.grpc.** - io.opencensus.** - io.perfmark.** - - - - - - - - - - + + + + org.apache.maven.plugins + maven-shade-plugin + 3.2.4 + + + package + + shade + + + + + + + + *:* + + META-INF/*.SF + META-INF/*.DSA + META-INF/*.RSA + + + + + + + + + diff --git a/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java b/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java new file mode 100644 index 000000000..423970d80 --- /dev/null +++ b/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java @@ -0,0 +1,67 @@ +package com.customer.app; + +import com.customer.util.DataframeUtils; +import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; +import com.google.cloud.secretmanager.v1.SecretVersionName; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; + +import java.io.IOException; + +public class PostgresToMySql { + + public static void main(String[] args) { + if (args.length != 6) { + System.err.println("Usage: PostgresToMySql "); + System.exit(1); + } + + String postgresTable = args[0]; + String mysqlTable = args[1]; + String postgresSecretId = args[2]; + String mysqlSecretId = args[3]; + String partitionColumn = args[4]; + String batchSize = args[5]; + + SparkSession spark = SparkSession.builder() + .appName("Postgres to MySQL JDBC Migration") + .getOrCreate(); + + String postgresUrl = getSecret(postgresSecretId); + String mysqlUrl = getSecret(mysqlSecretId); + + // Read data from Postgres in parallel + Dataset df = spark.read() + .format("jdbc") + .option("url", postgresUrl) + .option("dbtable", postgresTable) + .option("partitionColumn", partitionColumn) + .option("lowerBound", "1") + .option("upperBound", "1000") // These bounds should be configured based on data + .option("numPartitions", "4") + .load(); + + // Add insertion time column + Dataset transformedDf = DataframeUtils.addInsertionTimeColumn(df); + + // Write data to MySQL in batches + transformedDf.write() + .format("jdbc") + .option("url", mysqlUrl) + .option("dbtable", mysqlTable) + .option("batchsize", batchSize) + .save(); + + spark.stop(); + } + + private static String getSecret(String secretId) { + try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { + SecretVersionName secretVersionName = SecretVersionName.of("your-gcp-project-id", secretId, "latest"); + return client.accessSecretVersion(secretVersionName).getPayload().getData().toStringUtf8(); + } catch (IOException e) { + throw new RuntimeException("Failed to retrieve secret", e); + } + } +} From 02956bf6473d0048a96abb82d5f68e109bb9429d Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Sat, 24 Jan 2026 01:48:08 +0000 Subject: [PATCH 12/15] Working Java JDBC to JDBC --- gemini-codegen/java/migrateJdbcToJdbc.md | 13 ++- gemini-codegen/java/pom.xml | 71 +++++++++++--- .../com/customer/app/PostgresToMySql.java | 14 ++- .../java/com/customer/util/JdbcToJdbc.java | 92 ------------------- 4 files changed, 81 insertions(+), 109 deletions(-) delete mode 100644 gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java diff --git a/gemini-codegen/java/migrateJdbcToJdbc.md b/gemini-codegen/java/migrateJdbcToJdbc.md index fbcc561f5..252242504 100644 --- a/gemini-codegen/java/migrateJdbcToJdbc.md +++ b/gemini-codegen/java/migrateJdbcToJdbc.md @@ -1,7 +1,18 @@ # Migrating Data from Postgres to MySQL using a Serverless Spark Job This document provides instructions on how to run a Spark job to migrate data from a Postgres database to a MySQL database using a serverless Spark environment like Google Cloud Dataproc Serverless. - +--- +The text between the line above and line below was written by a human. The rest of the document was created by Gemini. The initial prompt to Gemini was: +``` +Create a Spark job in Java to to migrate data from a table in a Postgres database to a table in MySQL, both accessible via JDBC; The JDBC URL strings are stored in Google secret manager. The URL string includes the username and password. Read and write data in parallel based on partitioning information that is provided; While writing, write data in batches for efficiency. Use the addInsertionTimeColumn to add a column to the data before writing it to MySQL destination table. Provide instructions to run this job on serverless spark in migrateJdbcToJdbc.md Provide a summary of the session in migrationREADME.md +``` +Gemini generated the Java app, specifically the file `PostgresToMySql.java` and the README file. Changes were required to the generated code, including a)specifying drivers for Postgres and MySQL, b)adjusting partition bounds and number of partitions, converting a string argument to integer. At the time of this writing the generated `pom.xml` needed to be updated manually. Serverless for Apache Spark does not require a VPC subnet and the jdbc drivers are in the shaded Jar. The working gcloud command is: +``` +gcloud dataproc batches submit spark --class=com.customer.app.PostgresToMySql \ + --jars=/postgres-to-mysql-migration-1.0-SNAPSHOT.jar -- \ + +``` +--- ## Prerequisites 1. **Google Cloud SDK:** Make sure you have the Google Cloud SDK installed and configured on your local machine. diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index 86bd578dd..9f505e632 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -9,41 +9,52 @@ 1.0-SNAPSHOT - 1.8 - 1.8 - 3.2.0 + 11 + 3.5.7 + + + + com.google.cloud + libraries-bom + 26.73.0 + pom + import + + + org.apache.spark - spark-core_2.12 + spark-core_2.13 ${spark.version} + provided org.apache.spark - spark-sql_2.12 + spark-sql_2.13 ${spark.version} + provided org.postgresql postgresql - 42.3.3 + 42.7.8 - mysql - mysql-connector-java - 8.0.28 + com.mysql + mysql-connector-j + 9.5.0 com.google.cloud google-cloud-secretmanager - 2.0.0 @@ -52,7 +63,7 @@ org.apache.maven.plugins maven-shade-plugin - 3.2.4 + 3.6.1 package @@ -63,6 +74,16 @@ + + META-INF/DEPENDENCIES + + + false + + + + com.customer.app.PostgresToMySql + @@ -71,9 +92,37 @@ META-INF/*.SF META-INF/*.DSA META-INF/*.RSA + module-info.class + META-INF/versions/*/module-info.class + + + + + com.google + com.customer.shaded.com.google + + + + + io.grpc.netty.shaded + com.customer.shaded.io.grpc.netty.shaded + + + io + com.customer.shaded.io + + io.grpc.** + io.opencensus.** + io.perfmark.** + + + + diff --git a/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java b/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java index 423970d80..8854cbeba 100644 --- a/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java +++ b/gemini-codegen/java/src/main/java/com/customer/app/PostgresToMySql.java @@ -5,6 +5,7 @@ import com.google.cloud.secretmanager.v1.SecretVersionName; import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Row; +import org.apache.spark.sql.SaveMode; import org.apache.spark.sql.SparkSession; import java.io.IOException; @@ -37,9 +38,10 @@ public static void main(String[] args) { .option("url", postgresUrl) .option("dbtable", postgresTable) .option("partitionColumn", partitionColumn) - .option("lowerBound", "1") - .option("upperBound", "1000") // These bounds should be configured based on data - .option("numPartitions", "4") + .option("lowerBound", "2012-01-03") + .option("upperBound", "2025-10-31") // These bounds should be configured based on data + .option("numPartitions", "1") + .option("driver", "org.postgresql.Driver") .load(); // Add insertion time column @@ -50,7 +52,9 @@ public static void main(String[] args) { .format("jdbc") .option("url", mysqlUrl) .option("dbtable", mysqlTable) - .option("batchsize", batchSize) + .option("batchsize", Integer.parseInt(batchSize)) + .option("driver", "com.mysql.cj.jdbc.Driver") + .mode(SaveMode.Overwrite) .save(); spark.stop(); @@ -58,7 +62,7 @@ public static void main(String[] args) { private static String getSecret(String secretId) { try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { - SecretVersionName secretVersionName = SecretVersionName.of("your-gcp-project-id", secretId, "latest"); + SecretVersionName secretVersionName = SecretVersionName.of("dataproc-templates", secretId, "latest"); return client.accessSecretVersion(secretVersionName).getPayload().getData().toStringUtf8(); } catch (IOException e) { throw new RuntimeException("Failed to retrieve secret", e); diff --git a/gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java b/gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java deleted file mode 100644 index f58a76dbf..000000000 --- a/gemini-codegen/java/src/main/java/com/customer/util/JdbcToJdbc.java +++ /dev/null @@ -1,92 +0,0 @@ -/* - * Copyright (C) 2023 Google Inc. - * - * 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.customer.util; - -import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; -import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; -import com.google.cloud.secretmanager.v1.SecretVersionName; -import org.apache.spark.sql.Dataset; -import org.apache.spark.sql.Row; -import org.apache.spark.sql.SaveMode; -import org.apache.spark.sql.SparkSession; -import org.apache.spark.sql.functions; - -import java.io.IOException; -import java.io.StringReader; -import java.util.Properties; - -public class JdbcToJdbc { - - private static final String PROJECT_ID = "dataproc-templates"; - private static final String JDBC_INPUT_URL_SECRET_ID = "jdbc.input.url.secret.id"; - private static final String JDBC_INPUT_TABLE = "jdbc.input.table"; - private static final String JDBC_PARTITION_COLUMN = "jdbc.input.partition.column"; - private static final String JDBC_LOWER_BOUND = "jdbc.input.lower.bound"; - private static final String JDBC_UPPER_BOUND = "jdbc.input.upper.bound"; - private static final String JDBC_NUM_PARTITIONS = "jdbc.input.num.partitions"; - private static final String JDBC_OUTPUT_URL_SECRET_ID = "jdbc.output.url.secret.id"; - private static final String JDBC_OUTPUT_TABLE = "jdbc.output.table"; - private static final String JDBC_BATCH_SIZE = "jdbc.batch.size"; - - public static void main(String[] args) throws IOException { - Properties properties = new Properties(); - properties.load(new StringReader(String.join("\n", args))); - - SparkSession spark = SparkSession.builder() - .appName("JDBC to JDBC Data Migration") - .getOrCreate(); - - // Read from Postgres - Dataset sourceDf = spark.read() - .format("jdbc") - .option("url", getSecret(properties.getProperty(JDBC_INPUT_URL_SECRET_ID))) - .option("dbtable", properties.getProperty(JDBC_INPUT_TABLE)) - .option("partitionColumn", properties.getProperty(JDBC_PARTITION_COLUMN)) - .option("lowerBound", properties.getProperty(JDBC_LOWER_BOUND)) - .option("upperBound", properties.getProperty(JDBC_UPPER_BOUND)) - .option("numPartitions", properties.getProperty(JDBC_NUM_PARTITIONS)) - .option("driver", "org.postgresql.Driver") - .load(); - - // Add insertion time - Dataset transformedDf = sourceDf.withColumn("insertion_time", functions.current_timestamp()); - - // Write to MySQL - transformedDf.write() - .format("jdbc") - .option("url", getSecret(properties.getProperty(JDBC_OUTPUT_URL_SECRET_ID))) - .option("dbtable", properties.getProperty(JDBC_OUTPUT_TABLE)) - .option("batchsize", properties.getProperty(JDBC_BATCH_SIZE)) - .option("driver", "com.mysql.cj.jdbc.Driver") - .mode(SaveMode.Overwrite) - .save(); - } - - /** - * Retrieves a secret from Google Secret Manager. - * - * @param secretId The ID of the secret to retrieve. - * @return The secret value. - * @throws IOException If an I/O error occurs. - */ - private static String getSecret(String secretId) throws IOException { - try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { - SecretVersionName secretVersionName = SecretVersionName.of(PROJECT_ID, secretId, "latest"); - AccessSecretVersionResponse response = client.accessSecretVersion(secretVersionName); - return response.getPayload().getData().toStringUtf8(); - } - } -} From fbaa8f1fc23a935a99df0b301b3f16d4b6e5e3c8 Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Tue, 27 Jan 2026 18:59:33 +0000 Subject: [PATCH 13/15] Updated pom.xml to remove MANIFEST.MF warning. Updated ClassPathLister to list classes - not fully working --- gemini-codegen/java/pom.xml | 1 + .../com/customer/util/ClassPathLister.java | 67 ++++++++++++------- 2 files changed, 42 insertions(+), 26 deletions(-) diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index 9f505e632..96acbe0af 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -89,6 +89,7 @@ *:* + META-INF/MANIFEST.MF META-INF/*.SF META-INF/*.DSA META-INF/*.RSA diff --git a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java index 0cf3d62cf..871c55247 100644 --- a/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java +++ b/gemini-codegen/java/src/main/java/com/customer/util/ClassPathLister.java @@ -14,15 +14,17 @@ import java.net.HttpURLConnection; import java.net.URL; import java.util.Arrays; +import java.util.ArrayList; import java.util.Collections; import java.util.Enumeration; import java.util.HashMap; +import java.util.List; +import java.util.Map; import java.util.HashSet; import java.util.jar.Attributes; import java.util.jar.JarEntry; import java.util.jar.JarFile; import java.util.jar.Manifest; -import java.util.Map; import java.util.Properties; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -36,6 +38,7 @@ import org.apache.spark.sql.RowFactory; import org.apache.spark.sql.SaveMode; import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.types.ArrayType; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.StructType; import scala.collection.Seq; @@ -68,8 +71,8 @@ public static void main(String[] args) throws IOException { .add("fileName", DataTypes.StringType, true) .add("implementationTitle", DataTypes.StringType, true) .add("implementationVersion", DataTypes.StringType, true) - .add("artifactStringPOM", DataTypes.StringType, true) - .add("artifactStringSolr", DataTypes.StringType, true); + .add("GAV", DataTypes.StringType, true) + .add("classes", new ArrayType(DataTypes.StringType, true), true); Dataset df = spark.createDataFrame(Collections.emptyList(), schema); for (String entry : classPathEntries) { @@ -88,7 +91,7 @@ else if (file.isFile() && entry.toLowerCase().endsWith(".jar")) { } Map envVars = System.getenv(); - HashMap handyVars = new HashMap(envVars); + Map handyVars = new HashMap<>(envVars); Set valuableKeys = new HashSet<>(); Collections.addAll(valuableKeys, "DATAPROC_WORKLOAD_TYPE", "DATAPROC_IMAGE_VERSION", "DATAPROC_IMAGE_TYPE", "SPARK_SCALA_VERSION"); handyVars.keySet().retainAll(valuableKeys); @@ -163,8 +166,9 @@ public static Dataset listJar(SparkSession spark, File file, StructType sch implementationTitle = null; implementationVersion = null; } - String artifactStringPOM = null; + List defClasses = new ArrayList<>(); Enumeration entries = jarFile.entries(); + Map POMmap = new HashMap<> (); while (entries.hasMoreElements()) { JarEntry entry = entries.nextElement(); String name = entry.getName(); @@ -172,23 +176,35 @@ public static Dataset listJar(SparkSession spark, File file, StructType sch try (InputStream inputStream = jarFile.getInputStream(entry)) { Properties props = new Properties(); props.load(inputStream); - String groupId = props.getProperty("groupId"); - String artifactId = props.getProperty("artifactId"); - String version = props.getProperty("version"); - if (groupId != null && artifactId != null && version != null) { - artifactStringPOM = "" + groupId + "" + artifactId + "" + version + ""; + if (props.getProperty("groupId")!= null) { + POMmap.put("groupId", props.getProperty("groupId")); + POMmap.put("artifactId", props.getProperty("artifactId")); + POMmap.put("version", props.getProperty("version")); } } } + else if (name.endsWith(".class")) { + defClasses.add(entry.getName() + .replace("/", ".") + .replace(".class", "")); + } + } + ObjectMapper objectMapper = new ObjectMapper(); + if (POMmap.isEmpty()) { + Row row = RowFactory.create(file.getName(), implementationTitle, implementationVersion, + objectMapper.writeValueAsString(getGroupIdAndArtifactId(DigestUtils.sha1Hex(new FileInputStream(file)))), + defClasses); + return spark.createDataFrame(Collections.singletonList(row), schema); + } + else { + Row row = RowFactory.create(file.getName(), implementationTitle, implementationVersion, + objectMapper.writeValueAsString(POMmap),defClasses); + return spark.createDataFrame(Collections.singletonList(row), schema); } - String artifactStringSolr = getGroupIdAndArtifactId(DigestUtils.sha1Hex(new FileInputStream(file))); - Row row = RowFactory.create(file.getName(), implementationTitle, implementationVersion, artifactStringPOM, artifactStringSolr); - return spark.createDataFrame(Collections.singletonList(row), schema); - } catch (IOException e) { System.err.println("Error reading JAR file: " + file.getName()); e.printStackTrace(); - Row row = RowFactory.create(file.getName(), "Error", e.getMessage()); + Row row = RowFactory.create(file.getName(), "Error", e.getMessage(), null, null); return spark.createDataFrame(Collections.singletonList(row), schema); } @@ -201,7 +217,7 @@ public static Dataset listJar(SparkSession spark, File file, StructType sch * @return A string containing the groupId and artifactId, or null if not found. * @throws IOException If there is an issue with the HTTP connection. */ - public static String getGroupIdAndArtifactId(String sha1Checksum) throws IOException { + public static Map getGroupIdAndArtifactId(String sha1Checksum) throws IOException { final String MAVEN_SEARCH_API_URL = "https://search.maven.org/solrsearch/select?q=1:%s&rows=20&wt=json"; URL url = new URL(String.format(MAVEN_SEARCH_API_URL, sha1Checksum)); HttpURLConnection connection = (HttpURLConnection) url.openConnection(); @@ -225,12 +241,12 @@ public static String getGroupIdAndArtifactId(String sha1Checksum) throws IOExcep } } - private static String parseMavenResponse(String jsonResponse) { + private static Map parseMavenResponse(String jsonResponse) { // A simple parser to avoid adding a JSON library dependency. // This looks for "g":"...", "a":"..." and extracts the values. - String groupId = null; - String artifactId = null; - String version = null; + String groupId = ""; + String artifactId = ""; + String version = ""; String gPattern = "\"g\":\""; int gIndex = jsonResponse.indexOf(gPattern); @@ -260,12 +276,11 @@ private static String parseMavenResponse(String jsonResponse) { version = jsonResponse.substring(vStartIndex, vEndIndex); } } - - if (groupId != null && artifactId != null) { - return "" + groupId + "" + artifactId + "" + version + ""; - } - - return null; + return Map.of( + "groupId", groupId, + "artifactId", artifactId, + "version", version + ); } /** From 720de4c79a93ac1ef710ab2500e8687052b2db5e Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Tue, 27 Jan 2026 19:49:00 +0000 Subject: [PATCH 14/15] Migrate Deltalake to Iceberg generated by Gemini --- .../java/migrateDeltaLakeToIceberg.md | 75 +++++++++++++++++++ .../java/migrateDeltaLakeToIcebergREADME.md | 29 +++++++ gemini-codegen/java/pom.xml | 14 +++- .../com/customer/app/DeltaLakeToIceberg.java | 40 ++++++++++ 4 files changed, 156 insertions(+), 2 deletions(-) create mode 100644 gemini-codegen/java/migrateDeltaLakeToIceberg.md create mode 100644 gemini-codegen/java/migrateDeltaLakeToIcebergREADME.md create mode 100644 gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java diff --git a/gemini-codegen/java/migrateDeltaLakeToIceberg.md b/gemini-codegen/java/migrateDeltaLakeToIceberg.md new file mode 100644 index 000000000..b478a1603 --- /dev/null +++ b/gemini-codegen/java/migrateDeltaLakeToIceberg.md @@ -0,0 +1,75 @@ +# Migrating Data from Delta Lake to Iceberg + +This document provides instructions on how to run a Spark job to migrate data from a Delta Lake table in Google Cloud Storage (GCS) to an Iceberg table. + +## Prerequisites + +* A Dataproc Serverless cluster. +* A GCS bucket for the Iceberg warehouse. +* The JAR file for the Spark job. + +## Building the JAR + +To build the JAR file, run the following Maven command in your project's root directory: + +```bash +mvn clean package +``` + +This will create a JAR file in the `target` directory named `spark-delta-to-iceberg-migration-1.0-SNAPSHOT.jar`. + +## Running the Spark Job + +Use the following `gcloud` command to submit the Spark job to Dataproc Serverless: + +```bash +gcloud dataproc batches submit spark \ + --region= \ + --project= \ + --batch=delta-to-iceberg-migration \ + --class=com.customer.app.DeltaLakeToIceberg \ + --jars=target/spark-delta-to-iceberg-migration-1.0-SNAPSHOT.jar \ + -- \ + \ + \ + +``` + +### Arguments + +* ``: The region for the Dataproc Serverless cluster. +* ``: Your Google Cloud project ID. +* ``: The GCS path to the Delta Lake table (e.g., `gs:///delta-table`). +* ``: The name of the Iceberg table (e.g., `my_iceberg_table`). +* ``: The timestamp for time travel in the format `yyyy-MM-dd HH:mm:ss`. + +### Spark Configuration + +You will also need to provide additional configuration to your Spark job for Iceberg to work correctly. Create a file named `spark-properties.conf` with the following content: + +``` +spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions +spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog +spark.sql.catalog.spark_catalog.type=hive +spark.sql.catalog.spark_catalog.warehouse=gs:///iceberg-warehouse +``` + +Then, use the `--properties-file` flag when submitting your job: + +```bash +gcloud dataproc batches submit spark \ + --region= \ + --project= \ + --batch=delta-to-iceberg-migration \ + --class=com.customer.app.DeltaLakeToIceberg \ + --jars=target/spark-delta-to-iceberg-migration-1.0-SNAPSHOT.jar \ + --properties-file=spark-properties.conf \ + -- \ + \ + \ + +``` + +Replace `` with the name of your GCS bucket. + +``` diff --git a/gemini-codegen/java/migrateDeltaLakeToIcebergREADME.md b/gemini-codegen/java/migrateDeltaLakeToIcebergREADME.md new file mode 100644 index 000000000..c0c70c83a --- /dev/null +++ b/gemini-codegen/java/migrateDeltaLakeToIcebergREADME.md @@ -0,0 +1,29 @@ +# Summary of the Delta Lake to Iceberg Migration Session + +This document summarizes the steps taken to create and run a Spark job for migrating data from a Delta Lake table to an Iceberg table. + +## 1. Created the Spark Job + +A new Java class, `DeltaLakeToIceberg.java`, was created in `src/main/java/com/customer/app/`. This class contains the main logic for the Spark job, which: + +* Accepts the Delta Lake table path, Iceberg table name, and a timestamp as command-line arguments. +* Initializes a Spark session with the necessary Iceberg configurations. +* Reads the Delta Lake table from GCS using the provided timestamp for time travel. +* Writes the data to an Iceberg table. + +## 2. Updated the `pom.xml` + +The `pom.xml` file was updated to include the following dependencies: + +* `org.apache.iceberg:iceberg-spark-runtime-3.5_2.13:1.5.0` +* `io.delta:delta-spark_2.13:3.1.0` + +The `` was also changed to `spark-delta-to-iceberg-migration`, and the `` in the `maven-shade-plugin` was updated to `com.customer.app.DeltaLakeToIceberg`. + +## 3. Created `migrateDeltaLakeToIceberg.md` + +An instruction file, `migrateDeltaLakeToIceberg.md`, was created to provide a guide on how to build and run the Spark job on a Dataproc Serverless cluster. This file includes: + +* Instructions for building the JAR file using Maven. +* A `gcloud` command for submitting the Spark job. +* Details on the required command-line arguments and Spark configurations for Iceberg. diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index 96acbe0af..0a57766e3 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -5,7 +5,7 @@ 4.0.0 com.customer - postgres-to-mysql-migration + spark-delta-to-iceberg-migration 1.0-SNAPSHOT @@ -38,6 +38,16 @@ ${spark.version} provided + + org.apache.iceberg + iceberg-spark-runtime-3.5_2.13 + 1.5.0 + + + io.delta + delta-spark_2.13 + 3.1.0 + @@ -82,7 +92,7 @@ - com.customer.app.PostgresToMySql + com.customer.app.DeltaLakeToIceberg diff --git a/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java b/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java new file mode 100644 index 000000000..1a5d99b91 --- /dev/null +++ b/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java @@ -0,0 +1,40 @@ +package com.customer.app; + +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; + +public class DeltaLakeToIceberg { + + public static void main(String[] args) { + if (args.length != 3) { + System.err.println("Usage: DeltaLakeToIceberg "); + System.exit(1); + } + + String deltaTablePath = args[0]; + String icebergTableName = args[1]; + String timestamp = args[2]; + + SparkSession spark = SparkSession.builder() + .appName("Delta Lake to Iceberg Migration") + .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") + .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog") + .config("spark.sql.catalog.spark_catalog.type", "hive") + .getOrCreate(); + + // Read data from Delta Lake table using time travel + Dataset df = spark.read() + .format("delta") + .option("timestampAsOf", timestamp) + .load(deltaTablePath); + + // Write data to Iceberg table + df.write() + .format("iceberg") + .mode("overwrite") + .save(icebergTableName); + + spark.stop(); + } +} From 976f575e1052b18da609dc0c7f9cb769d264f15a Mon Sep 17 00:00:00 2001 From: Sundar Mudupalli <106282841+sundar-mudupalli-work@users.noreply.github.com> Date: Wed, 28 Jan 2026 17:56:45 +0000 Subject: [PATCH 15/15] Working DeltaLake to Iceberg migration --- gemini-codegen/java/migrateDeltaLakeToIceberg.md | 14 +++++++++++++- gemini-codegen/java/pom.xml | 15 +++++++++++---- .../java/com/customer/app/DeltaLakeToIceberg.java | 4 ++-- 3 files changed, 26 insertions(+), 7 deletions(-) diff --git a/gemini-codegen/java/migrateDeltaLakeToIceberg.md b/gemini-codegen/java/migrateDeltaLakeToIceberg.md index b478a1603..d5fa02194 100644 --- a/gemini-codegen/java/migrateDeltaLakeToIceberg.md +++ b/gemini-codegen/java/migrateDeltaLakeToIceberg.md @@ -1,7 +1,19 @@ # Migrating Data from Delta Lake to Iceberg This document provides instructions on how to run a Spark job to migrate data from a Delta Lake table in Google Cloud Storage (GCS) to an Iceberg table. - +--- +The text between the line above and line below was written by a human. The rest of the document was created by Gemini. The initial prompt to Gemini was: +``` +Create a Spark job in Java to migrate data from deltalake table in GCS to an Iceberg table. Use Timestamp based time travel of deltalake table. Provide instructions to run this job on Serverless Spark in migrateDeltaLakeToIceberg.md and provide a summary of the session in migrateDeltaLakeToIcebergREADME.md +``` +Gemini generated the Java app, specifically the file `DeltaLakeToIceberg.java` and the README file. Changes were required to the generated code, including a)specifying spark.sql.extensions and b) using `saveAsTable` instead of `save`. Gemini updated the existing `pom.xml`. A hive metastore URI is specified for the Serverless Spark job. The working gcloud command is: +``` +gcloud dataproc batches submit spark --version=2.2 --class=com.customer.app.DeltaLakeToIceberg \ + --jars= \ + --properties=spark.hadoop.hive.metastore.uris= \ + -- .. +``` +--- ## Prerequisites * A Dataproc Serverless cluster. diff --git a/gemini-codegen/java/pom.xml b/gemini-codegen/java/pom.xml index 0a57766e3..c5cd35ed7 100644 --- a/gemini-codegen/java/pom.xml +++ b/gemini-codegen/java/pom.xml @@ -82,15 +82,21 @@ - - + + + mozilla/public-suffix-list.txt + + META-INF/DEPENDENCIES + + NOTICE + false - + + com.customer.app.DeltaLakeToIceberg @@ -99,6 +105,7 @@ *:* + META-INF/maven/** META-INF/MANIFEST.MF META-INF/*.SF META-INF/*.DSA diff --git a/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java b/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java index 1a5d99b91..6faec161d 100644 --- a/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java +++ b/gemini-codegen/java/src/main/java/com/customer/app/DeltaLakeToIceberg.java @@ -18,7 +18,7 @@ public static void main(String[] args) { SparkSession spark = SparkSession.builder() .appName("Delta Lake to Iceberg Migration") - .config("spark.sql.extensions", "org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") + .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension,org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions") .config("spark.sql.catalog.spark_catalog", "org.apache.iceberg.spark.SparkSessionCatalog") .config("spark.sql.catalog.spark_catalog.type", "hive") .getOrCreate(); @@ -33,7 +33,7 @@ public static void main(String[] args) { df.write() .format("iceberg") .mode("overwrite") - .save(icebergTableName); + .saveAsTable(icebergTableName); spark.stop(); }