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://<output_data_bucket>/warehouse.

In the management console, select the folder.

Navigate to Managed Service for Apache Spark™.

Click the cluster name and select Jobs.

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.