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

Set up your 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 Metastore cluster	managed-metastore.integrationProvider: To enable the 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 Metastore cluster, create a security group named metastore-sg in the data-processing-network 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 the data-processing-network network. Add the following rules to the group:

   • 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
     • Source: Security group
     • Security group: Self

   • For outgoing HTTPS traffic:

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

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

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

8. Create a 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 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://<bucket_for_PySpark_job_output_data>/countries").saveAsTable("countries")
   

2. In <bucket_for_PySpark_job_source_code>, create a scripts folder 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, copy the following script to it and insert your infrastructure data in variables:

   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_interval='@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='A 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 be automatically scaled 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 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 a 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 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. The results are displayed 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 a folder named countries 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 Metastore cluster.
4. Make sure there are PySpark job logs in <bucket_for_collecting_Spark_logs>.

1. Service accounts.
2. Object Storage buckets
3. 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.