Exchanging data with Yandex Data Processing

With Yandex Data Processing, you can:

• Upload data from Managed Service for ClickHouse® to Spark DataFrame.
• Export data from Spark DataFrame to Managed Service for ClickHouse®.

If you no longer need the resources you created, delete them.

Getting startedGetting started

Prepare the infrastructure:

Upload data from Managed Service for ClickHouse®Upload data from Managed Service for ClickHouse®

Create a table in the Managed Service for ClickHouse® clusterCreate a table in the Managed Service for ClickHouse® cluster

1. Connect to the db1 database of the Managed Service for ClickHouse® cluster as user1.

2. Add test data to the database. As an example, a simple table is used with people's names and ages.

   1. Create a table:

      CREATE TABLE persons (
          `name` String,
          `age` UInt8) ENGINE = MergeTree ()
      ORDER BY
          `name`;
      

   2. Populate the table with data:

      INSERT INTO persons VALUES
         ('Anna', 19),
         ('Michael', 65),
         ('Alvar', 28),
         ('Lilith', 50),
         ('Max', 27),
         ('Jaimey', 34),
         ('Dmitry', 42),
         ('Qiang', 19),
         ('Augustyna', 20),
         ('Maria', 28);
      

   3. Check the result:

      SELECT * FROM persons;
      

Transfer the table from Managed Service for ClickHouse®Transfer the table from Managed Service for ClickHouse®

1. Prepare a script file:

   1. Create a local file named ch-to-dataproc.py and copy the following script to it:

      ch-to-dataproc.py

      from pyspark.sql import SparkSession
      
      # Creating a Spark session
      spark = SparkSession.builder.appName("ClickhouseDataproc").getOrCreate()
      
      # Setting the port and ClickHouse® cluster parameters
      jdbcPort = 8443
      jdbcHostname = "c-<ClickHouse®_cluster_ID>.rw.mdb.yandexcloud.net"
      jdbcDatabase = "db1"
      jdbcUrl = f"jdbc:clickhouse://{jdbcHostname}:{jdbcPort}/{jdbcDatabase}?ssl=true"
      
      # Transferring the persons table from ClickHouse® to DataFrame
      df = spark.read.format("jdbc") \
      .option("url", jdbcUrl) \
      .option("user","user1") \
      .option("password","<password_of_user1>") \
      .option("dbtable","persons") \
      .load()
      
      # Transferring the DataFrame to the bucket for checking
      df.repartition(1).write.mode("overwrite") \
      .csv(path='s3a://<output_bucket_name>/csv', header=True, sep=',')
      

   2. Specify the following in the script:

      • Managed Service for ClickHouse® cluster ID.
      • Password of user1.
      • Output bucket name.

   3. Create a scripts folder in the input bucket and upload the ch-to-dataproc.py file to it.

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

3. Wait for the job to complete and make sure the output bucket's csv folder contains the source table.

Export data to Managed Service for ClickHouse®Export data to Managed Service for ClickHouse®

1. Prepare a script file:

   1. Create a local file named dataproc-to-ch.py and copy the following script to it:

      dataproc-to-ch.py

      from pyspark.sql import SparkSession
      from pyspark.sql.types import *
      
      # Creating a Spark session
      spark = SparkSession.builder.appName("DataprocClickhouse").getOrCreate()
      
      # Creating a data schema
      schema = StructType([StructField('name', StringType(), True),
      StructField('age', IntegerType(), True)])
      
      # Creating a DataFrame
      df = spark.createDataFrame([('Alim', 19),
                                  ('Fred' ,65),
                                  ('Guanmin' , 28),
                                  ('Till', 60),
                                  ('Almagul', 27),
                                  ('Mary', 34),
                                  ('Dmitry', 42)], schema)
      
      # Specifying the port and ClickHouse® cluster parameters
      jdbcPort = 8443
      jdbcHostname = "c-<ClickHouse®_cluster_ID>.rw.mdb.yandexcloud.net"
      jdbcDatabase = "db1"
      jdbcUrl = f"jdbc:clickhouse://{jdbcHostname}:{jdbcPort}/{jdbcDatabase}?ssl=true"
      
      # Transferring the DataFrame to ClickHouse®
      df.write.format("jdbc") \
      .mode("error") \
      .option("url", jdbcUrl) \
      .option("dbtable", "people") \
      .option("createTableOptions", "ENGINE = MergeTree() ORDER BY age") \
      .option("user","user1") \
      .option("password","<ClickHouse®_database_password>") \
      .save()
      

   2. Specify the following in the script:

      • Managed Service for ClickHouse® cluster ID.
      • Password of user1.

   3. Create a scripts folder in the input bucket and upload the dataproc-to-ch.py file to it.

2. Create a PySpark job by specifying the path to the script file (s3a://<input bucket name>/scripts/dataproc-to-ch.py) in the Main python file field.

3. Wait for the job to complete and make sure the data has been transferred to Managed Service for ClickHouse®:

   1. Connect to the db1 database of the Managed Service for ClickHouse® cluster as user1.

   2. Run the following query:

      SELECT * FROM people;
      

   If the import is successful, the response will contain a table with the data.

Delete the resources you createdDelete the resources you created

Some resources are not free of charge. To avoid paying for them, delete the resources you no longer need:

ClickHouse® is a registered trademark of ClickHouse, Inc.