Shared use of Yandex Data Processing tables through Metastore

Create a service account named dataproc-s3-sa and assign the dataproc.agent and dataproc.provisioner roles to it.

In Object Storage, create buckets and configure access to them:

1. Create a bucket for the input data and grant the cluster service account READ permissions for this bucket.
2. Create a bucket for the processing output and grant the cluster service account READ and WRITE permissions for this bucket.

Create a cloud network named dataproc-network.

In the dataproc-network network, create a subnet in any availability zone.

Set up a NAT gateway for the subnet you created.

Create two Yandex Data Processing clusters named dataproc-source and dataproc-target in any suitable host configuration with the following settings:

• Services:
  • SPARK
  • YARN
• Service account: dataproc-sa.
• Properties: spark:spark.sql.hive.metastore.sharedPrefixes with the com.amazonaws,ru.yandex.cloud value. Required for PySpark jobs and integration with Metastore.
• Bucket name: Bucket you created for output data.
• Network: dataproc-network.

If the cloud network uses security groups, add the following rule for outgoing traffic to the Yandex Data Processing cluster security group:

• Port range: 9083.
• Protocol: Any (Any).
• Source: CIDR.
• CIDR blocks: 0.0.0.0/0.

Prepare a script file:

1. Create a local file named create-table.py and copy the following script to it:

   create-table.py

   from pyspark.sql.types import *
   from pyspark.sql import SparkSession
   
   # Creating a Spark session
   spark = SparkSession.builder \
       .appName("create-table") \
       .enableHiveSupport() \
       .getOrCreate()
   
   # Creating a data schema
   schema = StructType([StructField('Name', StringType(), True),
   StructField('Capital', StringType(), True),
   StructField('Area', IntegerType(), True),
   StructField('Population', IntegerType(), True)])
   
   # Creating a dataframe
   df = spark.createDataFrame([('Australia', 'Canberra', 7686850, 19731984), ('Austria', 'Vienna', 83855, 7700000)], schema)
   
   # Writing the dataframe to a bucket as a _countries_ table
   df.write.mode("overwrite").option("path","s3a://<output_bucket_name>/countries").saveAsTable("countries")
   

2. In the script, specify the name of the output bucket where the file with the countries table will be saved.

3. In the input bucket, create a folder named scripts and upload the create-table.py file to it.

Create a PySpark job by specifying the path to the script file in the Main python file field: s3a://<input_bucket_name>/scripts/create-table.py.

Wait for the job to complete and make sure the output bucket's countries folder contains the part-00000-... file.

Prepare a script file:

1. Create a local file named obtain-table.py and copy the following script to it:

   obtain-table.py

   from pyspark.sql import SparkSession
   
   # Creating a Spark session
   spark = SparkSession.builder \
       .appName("obtain-table") \
       .enableHiveSupport() \
       .getOrCreate()
   
   spark.catalog.listDatabases()
   
   # Getting the _countries_ table metadata from Metastore
   df = spark.sql("describe extended countries")
   
   # Requesting data from the _countries_ table
   df = spark.sql("select * from countries")
   
   # Testing table-to-bucket transfer
   df.repartition(1).write.csv(path='s3a://<output_bucket_name>/csv', header=True, sep=',')
   

2. In the script, specify the name of the output bucket the CSV file with the countries table will be saved to.

3. Upload the obtain-table.py file to the source data bucket's scripts folder.

Create a PySpark job by specifying the path to the script file in the Main python file field: s3a://<input_bucket_name>/scripts/obtain-table.py.

Wait for the job to complete and make sure the output bucket contains the csv folder with a table in CSV format.