Hello folks!<\/p>\n
We recently received a case letting us know that Dataproc 2.1.1 was unable to write to a BigQuery table with a column of type JSON. Although the BigQuery connector for Spark has had support for JSON columns since 0.28.0, the Dataproc images on the 2.1 line still cannot create tables with JSON columns or write to existing tables with JSON columns.<\/p>\n
The customer has graciously granted permission to share the code we developed to allow this operation. So if you are interested in working with JSON column tables on Dataproc 2.1 please continue reading!<\/p>\n
Use the following gcloud command to create your single-node dataproc cluster:<\/p>\n
IMAGE_VERSION=2.1.1-debian11\nREGION=us-west1\nZONE=${REGION}-a\nCLUSTER_NAME=pick-a-cluster-name\ngcloud dataproc clusters create ${CLUSTER_NAME} \\\n --region ${REGION} \\\n --zone ${ZONE} \\\n --single-node \\\n --master-machine-type n1-standard-4 \\\n --master-boot-disk-type pd-ssd \\\n --master-boot-disk-size 50 \\\n --image-version ${IMAGE_VERSION} \\\n --max-idle=90m \\\n --enable-component-gateway \\\n --scopes 'https:\/\/www.googleapis.com\/auth\/cloud-platform'\n<\/pre>\nThe following file is the Scala code used to write JSON structured data to a BigQuery table using Spark. The file following this one can be executed from your single-node Dataproc cluster.<\/p>\n
Main.scala<\/a><\/p>\nimport org.apache.spark.sql.functions.col\nimport org.apache.spark.sql.types.{Metadata, StringType, StructField, StructType}\nimport org.apache.spark.sql.{Row, SaveMode, SparkSession}\nimport org.apache.spark.sql.avro\nimport org.apache.avro.specific\n\n val env = \"x\"\n val my_bucket = \"cjac-docker-on-yarn\"\n val my_table = \"dataset.testavro2\"\n val spark = env match {\n case \"local\" =>\n SparkSession\n .builder()\n .config(\"temporaryGcsBucket\", my_bucket)\n .master(\"local\")\n .appName(\"isssue_115574\")\n .getOrCreate()\n case _ =>\n SparkSession\n .builder()\n .config(\"temporaryGcsBucket\", my_bucket)\n .appName(\"isssue_115574\")\n .getOrCreate()\n }\n\n \/\/ create DF with some data\n val someData = Seq(\n Row(\"\"\"{\"name\":\"name1\", \"age\": 10 }\"\"\", \"id1\"),\n Row(\"\"\"{\"name\":\"name2\", \"age\": 20 }\"\"\", \"id2\")\n )\n val schema = StructType(\n Seq(\n StructField(\"user_age\", StringType, true),\n StructField(\"id\", StringType, true)\n )\n )\n\n val avroFileName = s\"gs:\/\/${my_bucket}\/issue_115574\/someData.avro\"\n \n val someDF = spark.createDataFrame(spark.sparkContext.parallelize(someData), schema)\n someDF.write.format(\"avro\").mode(\"overwrite\").save(avroFileName)\n\n val avroDF = spark.read.format(\"avro\").load(avroFileName)\n \/\/ set metadata\n val dfJSON = avroDF\n .withColumn(\"user_age_no_metadata\", col(\"user_age\"))\n .withMetadata(\"user_age\", Metadata.fromJson(\"\"\"{\"sqlType\":\"JSON\"}\"\"\"))\n\n dfJSON.show()\n dfJSON.printSchema\n\n \/\/ write to BigQuery\n dfJSON.write.format(\"bigquery\")\n .mode(SaveMode.Overwrite)\n .option(\"writeMethod\", \"indirect\")\n .option(\"intermediateFormat\", \"avro\")\n .option(\"useAvroLogicalTypes\", \"true\")\n .option(\"table\", my_table)\n .save()\n\n\n<\/pre>\nrepro.sh<\/a>:<\/p>\n#!\/bin\/bash\n\nPROJECT_ID=set-yours-here\nDATASET_NAME=dataset\nTABLE_NAME=testavro2\n\n# We have to remove all of the existing spark bigquery jars from the local\n# filesystem, as we will be using the symbols from the\n# spark-3.3-bigquery-0.30.0.jar below. Having existing jar files on the\n# local filesystem will result in those symbols having higher precedence\n# than the one loaded with the spark-shell.\nsudo find \/usr -name 'spark*bigquery*jar' -delete\n\n# Remove the table from the bigquery dataset if it exists\nbq rm -f -t $PROJECT_ID:$DATASET_NAME.$TABLE_NAME\n\n# Create the table with a JSON type column\nbq mk --table $PROJECT_ID:$DATASET_NAME.$TABLE_NAME \\\n user_age:JSON,id:STRING,user_age_no_metadata:STRING\n\n# Load the example Main.scala \nspark-shell -i Main.scala \\\n --jars \/usr\/lib\/spark\/external\/spark-avro.jar,gs:\/\/spark-lib\/bigquery\/spark-3.3-bigquery-0.30.0.jar\n\n# Show the table schema when we use `bq mk --table` and then load the avro\nbq query --use_legacy_sql=false \\\n \"SELECT ddl FROM $DATASET_NAME.INFORMATION_SCHEMA.TABLES where table_name='$TABLE_NAME'\"\n\n# Remove the table so that we can see that the table is created should it not exist\nbq rm -f -t $PROJECT_ID:$DATASET_NAME.$TABLE_NAME\n\n# Dynamically generate a DataFrame, store it to avro, load that avro,\n# and write the avro to BigQuery, creating the table if it does not already exist\n\nspark-shell -i Main.scala \\\n --jars \/usr\/lib\/spark\/external\/spark-avro.jar,gs:\/\/spark-lib\/bigquery\/spark-3.3-bigquery-0.30.0.jar\n\n# Show that the table schema does not differ from one created with a bq mk --table\nbq query --use_legacy_sql=false \\\n \"SELECT ddl FROM $DATASET_NAME.INFORMATION_SCHEMA.TABLES where table_name='$TABLE_NAME'\"\n<\/pre>\nGoogle BigQuery has supported JSON data<\/a> since October of 2022<\/a>, but until now, it has not been possible, on generally available Dataproc clusters, to interact with these columns using the Spark BigQuery Connector.<\/p>\nJSON column type support was introduced in spark-bigquery-connector release 0.28.0<\/a>.<\/p>\n\n