Working with Yandex Object Storage from a PySpark job using an Apache Hive™ Metastore cluster

Create a service account named spark-agent for the Yandex Managed Service for Apache Spark™ cluster and assign it the managed-spark.integrationProvider role to enable the Yandex Managed Service for Apache Spark™ cluster to interact with other resources.

Create a service account named metastore-agent and assign it the managed-metastore.integrationProvider and storage.uploader roles to enable your Apache Hive™ Metastore cluster to interact with other resources and export metadata to the Object Storage bucket.

Create buckets:

• One for the PySpark job source code.
• One for output data.

Grant permissions to the spark-agent service account for the created buckets:

• Bucket for the PySpark job source code: READ.
• Bucket for output data: READ and WRITE.

Grant the READ and WRITE permissions for the output bucket to the metastore-agent service account.

Create a cloud network named integration-network.

This will automatically create three subnets in different availability zones.

For the Yandex Managed Service for Apache Spark™ cluster, create a security group named spark-sg in integration-network. Add the following rule to it:

• For outgoing traffic, to allow Yandex Managed Service for Apache Spark™ cluster connections to Apache Hive™ Metastore:

  • Port range: 9083
  • Protocol: Any
  • Destination: CIDR
  • CIDR blocks: 0.0.0.0/0

For the Apache Hive™ Metastore cluster, create a security group named metastore-sg in integration-network. Add the following rules to it:

• For incoming client traffic:

  • Port range: 30000-32767
  • Protocol: Any
  • Source: CIDR
  • CIDR blocks: 0.0.0.0/0

• For incoming load balancer traffic:

  • Port range: 10256
  • Protocol: Any
  • Source: Load balancer health checks

Create a Apache Hive™ Metastore cluster with the following parameters:

• Service account: metastore-agent.
• Version: 3.1.
• Network: integration-network.
• Subnet: integration-network-ru-central1-a.
• Security groups: metastore-sg.

Create a Yandex Managed Service for Apache Spark™ cluster with the following parameters:

• Service account: spark-agent.
• Network: integration-network.
• Subnet: integration-network-ru-central1-a.
• Security groups: spark-sg.
• Metastore: Apache Hive™ Metastore cluster you created earlier.

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

import random
import sys
from pyspark.sql import SparkSession


def prepare_table(spark, database, table):
    create_database_sql = "create database if not exists {database}"
    create_table_sql = """
    create table if not exists {database}.{table} (
        id int,
        value double
    )
    """
    truncate_table_sql = "truncate table {database}.{table}"

    spark.sql(create_database_sql.format(database=database))
    spark.sql(create_table_sql.format(database=database, table=table))
    spark.sql(truncate_table_sql.format(database=database, table=table))


def write_data(spark, database, table):
    data = [(i, random.random()) for i in range(100_000)]
    # Creating a dataframe
    df = spark.createDataFrame(data, schema=['id', 'value'])
    table_full_name = "{database}.{table}".format(database=database, table=table)
    df.write.mode('overwrite').format('json').saveAsTable(table_full_name)


def main():
    # Creating a Spark session
    spark = (
        SparkSession
        .builder
        .appName('job-create-table')
        .enableHiveSupport()
        .config('spark.sql.warehouse.dir', sys.argv[1])
        .getOrCreate()
    )
    database, table = 'database_1', 'table_1'
    prepare_table(spark, database, table)
    write_data(spark, database, table)


if __name__ == '__main__':
    if len(sys.argv) != 2:
        print("Usage: job-create-table s3a://<bucket>/<folder>", file=sys.stderr)
        sys.exit(-1)
    main()

In the source code bucket, create a folder named scripts and upload the job-create-table.py file to this folder.

In the output bucket, create the warehouse folder to load data from database_1 to.

Create a job with the following settings:

• Job type: PySpark
• Main python file: s3a://<source_code_bucket>/scripts/job-create-table.py
• Arguments: s3a://<bucket_for_output_data>/warehouse

Navigate to the folder dashboard and select Managed Service for Apache Spark™.

Click the name of your cluster and select the Jobs tab.

Wait for the PySpark job you created to change its status to Done.

Make sure the file with data from database_1 appears in the warehouse folder in your output data bucket.

Make sure the Apache Hive™ Metastore cluster has the metadata on database_1:

1. Export the metadata from the Apache Hive™ Metastore cluster to the output bucket.
2. Download the metadata file and make sure it mentions database_1.