Importing data from Yandex Object Storage, processing and exporting it to Yandex Managed Service for ClickHouse®
This tutorial is based on a Data Stories |
|
In this example, we use two CSV tables which we will merge into a single one, import it to Parquet format, and transfer to Managed Service for ClickHouse®.
Getting started
Prepare the infrastructure:
-
Create a service account named
dataproc-s3-sa
and assign thedataproc.agent
anddataproc.provisioner
roles to it. -
In Object Storage, create buckets and configure access to them:
- Create a bucket for the input data and grant the cluster service account
READ
permissions for this bucket. - Create a bucket for the processing output and grant the cluster service account
READ and WRITE
permissions for this bucket.
- Create a bucket for the input data and grant the cluster service account
-
Create a cloud network named
dataproc-network
. -
Create a subnet in any availability zone in
dataproc-network
. -
Set up a NAT gateway for the subnet you created.
-
In
dataproc-network
, create a security group nameddataproc-sg
and add the following rules to it:-
One rule for inbound and another one for outbound service traffic:
- Port range:
0-65535
- Protocol:
Any
- Source / Destination name:
Security group
- Security group:
Current
(Self
)
- Port range:
-
Rule for outgoing HTTPS traffic:
- Port range:
443
- Protocol:
TCP
- Destination name:
CIDR
- CIDR blocks:
0.0.0.0/0
- Port range:
-
Rule for outgoing TCP traffic on port 8443 to access ClickHouse®:
- Port range:
8443
- Protocol:
TCP
- Destination name:
CIDR
- CIDR blocks:
0.0.0.0/0
- Port range:
-
-
Create a Yandex Data Processing cluster in any suitable host configuration with the following settings:
- Services:
SPARK
YARN
HDFS
- Service account:
dataproc-sa
. - Bucket name: Bucket you created for output data.
- Network:
dataproc-network
. - Security groups:
dataproc-sg
. - UI Proxy: Enabled.
- Services:
-
Create a Managed Service for ClickHouse® cluster in any suitable configuration with the following settings:
- With public access to cluster hosts.
- DB name:
db1
. - Username:
user1
.
-
If you do not have Terraform yet, install it.
-
Get the authentication credentials. You can add them to environment variables or specify them later in the provider configuration file.
-
Configure and initialize a provider. There is no need to create a provider configuration file manually, you can download it
. -
Place the configuration file in a separate working directory and specify the parameter values. If you did not add the authentication credentials to environment variables, specify them in the configuration file.
-
Download the s3-dataproc-ch.tf
configuration file to the same working directory.This file describes:
- Network.
- Subnet.
- NAT gateway and route table required for Yandex Data Processing.
- Security groups required for the Yandex Data Processing and Managed Service for ClickHouse® clusters.
- Service account required for the Yandex Data Processing cluster to run.
- Service account required to create buckets in Object Storage.
- Buckets for input and output data.
- Yandex Data Processing cluster.
- Managed Service for ClickHouse® cluster.
-
Specify the following in the
s3-dataproc-ch.tf
file:folder_id
: Cloud folder ID, same as in the provider settings.input-bucket
: Name of the bucket for input data.output-bucket
: Name of the bucket for output data.dp_ssh_key
: Absolute path to the public key for the Yandex Data Processing cluster. For more information, see Connecting to a Yandex Data Processing host via SSH.ch_password
: ClickHouse® user password.
-
Make sure the Terraform configuration files are correct using this command:
terraform validate
If there are any errors in the configuration files, Terraform will point them out.
-
Create the required infrastructure:
-
Run the command to view planned changes:
terraform plan
If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.
-
If you are happy with the planned changes, apply them:
-
Run the command:
terraform apply
-
Confirm the update of resources.
-
Wait for the operation to complete.
-
All the required resources will be created in the specified folder. You can check resource availability and their settings in the management console
. -
Prepare the test data
In this example, we use the following two CSV tables:
coords.csv
with vehicle geographic coordinates.sensors.csv
with information about the speed and other vehicle operating parameters.
To prepare the test data:
-
Copy the contents of the sample files below and save them locally in CSV format:
-
coords.csv
vehicle_id,latitude,longitude,altitude iv9a94th6rztooxh5ur2,55.70329032,37.65472196,427.5 022wsiz48h2ljxuz04x8,56.96149325,38.46541766,423.6 a7fbbqjws4zqw85f6jue,54.99296663,36.79063999,426.2 l7731117m6r6on4m633n,55.34740545,37.13175678,422.5 6f9q6ienc4qfpdwd9nef,56.69752218,38.38871530,428.3
-
sensors.csv
vehicle_id,speed,battery_voltage,cabin_temperature,fuel_level iv9a94th6rztooxh5ur2,0.0,25.5,17,5 022wsiz48h2ljxuz04x8,55.5,54.5,21,22 a7fbbqjws4zqw85f6jue,80.6,22.1,19,73 l7731117m6r6on4m633n,40.9,76.0,25,23 6f9q6ienc4qfpdwd9nef,64.8,90.8,21,32
-
-
Create the
csv
folder in the input bucket and upload the created CSV files to it.
Process your data in Yandex Data Processing
Merge the data from the two tables into one and upload it in Parquet format to the bucket you previously created for processing results:
-
Prepare a script file:
-
Create a local file named
join-tables.py
and copy the following script to it:join-tables.py
from pyspark.sql import SparkSession # Creating a Spark session spark = SparkSession.builder.appName("JoinExample").getOrCreate() # Reading the table from the coords.csv file coords_df = spark.read.csv("s3a://<input_bucket_name>/csv/coords.csv", header=True) # Reading the table from the sensors.csv file sensors_df = spark.read.csv("s3a://<input_bucket_name>/csv/sensors.csv", header=True) # Joining the tables by the vehicle_id column joined_df = coords_df.join(sensors_df, on="vehicle_id", how="inner") # Saving the joined table to the bucket in Parquet format joined_df.write.parquet("s3a://<output_bucket_name>/parquet/")
-
Specify the following in the script:
- Name of the input bucket that stores the original CSV tables.
- Name of the output bucket where the Parquet file with the merged data will be saved.
-
Create the
scripts
folder in the input bucket and upload thejoin-tables.py
file to it.
-
-
Create a PySpark job by specifying the path to the script file (
s3a://<input_bucket_name>/scripts/join-tables.py
) in the Main python file field. -
Wait for the job to complete and make sure the output bucket's
parquet
folder contains thepart-00000-***
Parquet file.
Note
You can view the job logs and search data in them using Yandex Cloud Logging. For more information, see Working with logs.
Export your data to ClickHouse®
Transfer the joined table from Object Storage to ClickHouse®:
-
Prepare a script file:
-
Create a local file named
parquet-to-ch.py
and copy the following script to it:parquet-to-ch.py
from pyspark.sql import SparkSession # Creating a Spark session spark = SparkSession.builder.appName("ParquetClickhouse").getOrCreate() # Reading the data from the Parquet file parquetFile = spark.read.parquet("s3a://<output_bucket_name>/parquet/*.parquet") # Specifying the port and ClickHouse® cluster parameters jdbcPort = 8443 jdbcHostname = "c-<cluster_ID>.rw.mdb.yandexcloud.net" jdbcDatabase = "db1" jdbcUrl = f"jdbc:clickhouse://{jdbcHostname}:{jdbcPort}/{jdbcDatabase}?ssl=true" # Transferring the table from the Parquet file to a ClickHouse® table named "measurements" parquetFile.write.format("jdbc") \ .mode("error") \ .option("url", jdbcUrl) \ .option("dbtable", "measurements") \ .option("createTableOptions", "ENGINE = MergeTree() ORDER BY vehicle_id") \ .option("user","user1") \ .option("password","<ClickHouse®_user_password>") \ .save()
-
Specify the following in the script:
- Name of the bucket with the Parquet file.
- Managed Service for ClickHouse® cluster ID.
- ClickHouse® user password.
-
Upload the
parquet-to-ch.py
file to the input bucket'sscripts
folder.
-
-
Create a PySpark job by specifying the path to the script file (
s3a://<input_bucket_name>/scripts/parquet-to-ch.py
) in the Main python file field. -
Wait for the job to complete and make sure the joined table has been moved to the cluster:
-
Connect to the
db1
database of the Managed Service for ClickHouse® cluster asuser1
. -
Run the following query:
SELECT * FROM measurements;
If the data export is successful, the response will contain the joined table.
-
Delete the resources you created
Some resources are not free of charge. To avoid paying for them, delete the resources you no longer need:
-
Delete the objects from the buckets.
-
In the terminal window, go to the directory containing the infrastructure plan.
-
Delete the
s3-dataproc-ch.tf
configuration file. -
Make sure the Terraform configuration files are correct using this command:
terraform validate
If there are any errors in the configuration files, Terraform will point them out.
-
Confirm updating the resources.
-
Run the command to view planned changes:
terraform plan
If the resource configuration descriptions are correct, the terminal will display a list of the resources to modify and their parameters. This is a test step. No resources are updated.
-
If you are happy with the planned changes, apply them:
-
Run the command:
terraform apply
-
Confirm the update of resources.
-
Wait for the operation to complete.
-
All the resources described in the
s3-dataproc-ch.tf
configuration file will be deleted. -
ClickHouse® is a registered trademark of ClickHouse, Inc