Automating operations with Yandex Data Processing using Yandex Managed Service for Apache Airflow™

Set up the infrastructure:

1. Create service accounts with the following roles:

   Service account	Roles
   airflow-agent for an Apache Airflow™ cluster	dataproc.editor to manage a Yandex Data Processing cluster from a DAG. vpc.user to use the Yandex Virtual Private Cloud subnet in the Apache Airflow™ cluster. managed-airflow.integrationProvider to enable the Apache Airflow™ cluster to interact with other resources. iam.serviceAccounts.user to specify the data-processing-agent service account when creating a Yandex Data Processing cluster.
   metastore-agent for a Apache Hive™ Metastore cluster	managed-metastore.integrationProvider to enable the Apache Hive™ Metastore cluster to interact with other resources.
   data-processing-agent for a Yandex Data Processing cluster	dataproc.agent: To enable the service account to get info on cluster host states, jobs, and log groups. dataproc.provisioner: To enable the service account to work with an autoscaling instance group. This will enable subcluster autoscaling.

2. Create buckets:

   • <bucket_for_Managed_Airflow>
   • <bucket_for_PySpark_job_source_code>
   • <bucket_for_PySpark_job_output_data>
   • <bucket_for_collecting_Spark_logs>

   You need multiple buckets with different access permissions.

3. Grant permissions for the buckets as follows:

   • <bucket_for_Managed_Airflow>: READ permission to the airflow-agent service account.
   • <bucket_for_PySpark_job_source_code>: READ permission to the data-processing-agent service account.
   • <bucket_for_PySpark_job_output_data>: READ and WRITE permissions to the data-processing-agent and metastore-agent service accounts.
   • <bucket_for_collecting_Spark_logs>: READ and WRITE permissions to the data-processing-agent service account.

4. Create a cloud network named data-processing-network.

   This will automatically create three subnets in different availability zones.

5. Set up a NAT gateway for the data-processing-network-ru-central1-a subnet.

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

   • 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

7. For the Managed Service for Apache Airflow™ and Yandex Data Processing clusters, create a security group named airflow-sg in data-processing-network. Add the following rules to it:

   • For incoming service traffic:

     • Port range: 0-65535
     • Protocol: Any
     • Source: Security group
     • Security group: Self

   • For outgoing service traffic:

     • Port range: 0-65535
     • Protocol: Any
     • Destination: Security group
     • Security group: Self

   • For outgoing HTTPS traffic:

     • Port range: 443
     • Protocol: TCP
     • Destination: CIDR
     • CIDR blocks: 0.0.0.0/0

   • For outgoing traffic, to allow Yandex Data Processing cluster connections to Apache Hive™ Metastore:

     • Port range: 9083
     • Protocol: Any
     • Destination: Security group
     • Security group: metastore-sg (From list)

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

   • Service account: metastore-agent
   • Network: data-processing-network
   • Subnet: data-processing-network-ru-central1-a
   • Security group: metastore-sg

9. Create a Managed Service for Apache Airflow™ cluster with the following parameters:

   • Service account: airflow-agent
   • Availability zone: ru-central1-a
   • Network: data-processing-network
   • Subnet: data-processing-network-ru-central1-a
   • Security group: airflow-sg
   • Bucket name: <bucket_for_Managed_Airflow>

1. Create a local file named create-table.py and paste 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://<bucket_for_PySpark_job_output_data>/countries").saveAsTable("countries")
   

2. In <bucket_for_PySpark_job_source_code>, create a folder named scripts and upload the create-table.py file to it.

1. Create an SSH key. Save the public part of the key: you will need it to create a Yandex Data Processing cluster.

2. Create a local file named Data-Processing-DAG.py, paste the following script to it and substitute the variables with your infrastructure data:

   Data-Processing-DAG.py

   import uuid
   import datetime
   from airflow import DAG
   from airflow.utils.trigger_rule import TriggerRule
   from airflow.providers.yandex.operators.yandexcloud_dataproc import (
       DataprocCreateClusterOperator,
       DataprocCreatePysparkJobOperator,
       DataprocDeleteClusterOperator,
   )
   
   # Your infrastructure data
   YC_DP_AZ = 'ru-central1-a'
   YC_DP_SSH_PUBLIC_KEY = '<public_part_of_SSH_key>'
   YC_DP_SUBNET_ID = '<subnet_ID>'
   YC_DP_SA_ID = '<ID_of_data_processing_agent_service_account>'
   YC_DP_METASTORE_URI = '<IP_address>'
   YC_SOURCE_BUCKET = '<bucket_for_PySpark_job_source_code>'
   YC_DP_LOGS_BUCKET = '<bucket_for_collecting_Spark_logs>'
   
   # DAG settings
   with DAG(
           'DATA_INGEST',
           schedule='@hourly',
           tags=['data-processing-and-airflow'],
           start_date=datetime.datetime.now(),
           max_active_runs=1,
           catchup=False
   ) as ingest_dag:
       # Step 1: Creating a Yandex Data Proc cluster
       create_spark_cluster = DataprocCreateClusterOperator(
           task_id='dp-cluster-create-task',
           cluster_name=f'tmp-dp-{uuid.uuid4()}',
           cluster_description='Temporary cluster for running a PySpark job orchestrated by Managed Service for Apache Airflow™',
           ssh_public_keys=YC_DP_SSH_PUBLIC_KEY,
           service_account_id=YC_DP_SA_ID,
           subnet_id=YC_DP_SUBNET_ID,
           s3_bucket=YC_DP_LOGS_BUCKET,
           zone=YC_DP_AZ,
           cluster_image_version='2.1',
           masternode_resource_preset='s2.small',
           masternode_disk_type='network-ssd',
           masternode_disk_size=200,
           computenode_resource_preset='m2.large',
           computenode_disk_type='network-ssd',
           computenode_disk_size=200,
           computenode_count=2,
           computenode_max_hosts_count=5,  # The number of data-handling subclusters will automatically scale up if the load is high.
           services=['YARN', 'SPARK'],     # A lightweight cluster is being created.
           datanode_count=0,               # Without data storage subclusters.
           properties={                    # With a reference to a remote Apache Hive™ Metastore cluster.
               'spark:spark.hive.metastore.uris': f'thrift://{YC_DP_METASTORE_URI}:9083',
           },
       )
   
       # Step 2: Running a PySpark job
       poke_spark_processing = DataprocCreatePysparkJobOperator(
           task_id='dp-cluster-pyspark-task',
           main_python_file_uri=f's3a://{YC_SOURCE_BUCKET}/scripts/create-table.py',
       )
   
       # Step 3: Deleting the Yandex Data Processing cluster
       delete_spark_cluster = DataprocDeleteClusterOperator(
           task_id='dp-cluster-delete-task',
           trigger_rule=TriggerRule.ALL_DONE,
       )
   
       # Building a DAG based on the above steps
       create_spark_cluster >> poke_spark_processing >> delete_spark_cluster
   

   Where:

   • YC_DP_AZ: Availability zone for the Yandex Data Processing cluster.
   • YC_DP_SSH_PUBLIC_KEY: Public part of the SSH key for the Yandex Data Processing cluster.
   • YC_DP_SUBNET_ID: Subnet ID.
   • YC_DP_SA_ID: ID of the service account for Yandex Data Processing.
   • YC_DP_METASTORE_URI: IP address of the Apache Hive™ Metastore cluster.
   • YC_SOURCE_BUCKET: Bucket with the Python script for the PySpark job.
   • YC_DP_LOGS_BUCKET: Bucket for logs.

3. Upload the DAG to the Managed Service for Apache Airflow™ cluster: in <bucket_for_Managed_Airflow>, create a folder named dags and upload the Data-Processing-DAG.py file to it.

4. Open the Apache Airflow™ web interface.

5. Make sure the DAGs section now contains the new DAG named DATA_INGEST and tagged as data-processing-and-airflow.

   It may take a few minutes to upload a DAG file from the bucket.

6. To run the DAG, click  in the line with its name.

1. To monitor task execution results, click the DAG name. You can find the results in the Grid tab.
2. Wait until the status of all the three tasks in the DAG changes to Success. Simultaneously, you can check that a Yandex Data Processing cluster is being created, the PySpark job is running, and the same cluster is being deleted in the management console.
3. Make sure your <bucket_for_PySpark_job_output_data> now contains the countries folder with the part-00000-... file. The data from the created table is now stored in the Object Storage bucket and the table metadata is stored in the Apache Hive™ Metastore cluster.
4. Make sure there are PySpark job logs in <bucket_for_collecting_Spark_logs>.

1. Service accounts.
2. Object Storage buckets.
3. Apache Hive™ Metastore cluster.
4. Managed Service for Apache Airflow™ cluster.
5. Route table.
6. NAT gateway.
7. Security groups.
8. Cloud subnets created in data-processing-network by default.
9. Cloud network.