Shared use of Yandex Data Processing tables through Metastore

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

2. 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.

3. Create a cloud network named dataproc-network.

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

5. Set up a NAT gateway for the subnet you created.

6. 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.

7. 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.

1. 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.

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

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

1. 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.

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

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