Data Sync to Snowflake Using Confluent Kafka Connect: Part 1
Data Sync to Snowflake Using Confluent Kafka Connect: Part 1

Moving On-Prem Oracle Databases to Snowflake in Azure with Kafka Connect

In my most recent engagement, I was tasked with data synchronization between an on-premise Oracle database with Snowflake using Confluent Kafka. While there are many blogs that cover this topic, they don’t provide scenarios that deviate from the normal happy path and demonstrate how to overcome the deviations.

In this blog post, I will go over some of the hurdles we observed during the engagement and explain how we solved them.

The Scenario

The customer had an on-premise Oracle database. They wanted to synchronize the data into Snowflake in Azure and also wanted the ability to use the same data transport channel for future uses. They chose Confluent Kafka as the candidate for transferring the data from on-premise to cloud.

Here are some of the requirements that were mandated as part of the solution:

Develop a solution that could be adopted for multiple instances of Oracle databases.

Synchronization into Snowflake does not need to be real-time.

Adopt tools/libraries that are supplied by Confluent or the Confluent community before considering third-party tools.

Define a solution that is economical; preferably open-source and free.

The client wanted to adopt a pattern similar to this one below:

My Initial Thoughts Were…

The solution to this data synchronization did not seem complicated and appeared to be achievable. There are a number of implementations for this scenario as described in these articles:

The Hiccups

At first glance these solutions seemed workable, however, they did not end up suitable for the current client situation because of the following reasons:

1) Adoption Cycle

In the adoption cycle, we were in the early stages of Snowflake adoption. We still had to prove a use case and get the buy-in from the business for a complete adoption. For this reason, we wanted to limit and control the tech stack.

2) Timelines

There is a plethora of integration software that can achieve the solution we want. We did not have the necessary cycles, times, or resources in this project to evaluate new software vendors.

3) Enterprise Practices

Even if we had chosen a specific vendor (e.g. Attunity, Fivetran, Matillion, etc.), the timeline for getting contracts in place, negotiating the subscription/license, going through a security assessment, etc. was significant and didn’t fit the project timeline.

The Tactical Approach

So far we had Confluent and Snowflake already spun up, contracted for, and ready to go. We wanted to get started with what we already had, so we decided to develop our solution, even if it happened to be tactical, with the following stack:

Confluent JDBC connector: To pull data from the on-premise Oracle database.

Confluent Connect: The framework to connect to Kafka with external systems, including databases. It provided a solution to use community connectors and slash development cycles.

Snowflake Connector for Kafka: To push data into Snowflake.

The following diagram depicts the approach we wanted to implement:

In the following set of topics, I will highlight some of the hurdles that we faced and discuss how we solved these. Though some of the solutions are tactical in nature, it gave us an edge to proceed with achieving our goals in the defined timeframe.

NOTE: I am using local Confluent and Oracle instances with Docker containers to capture some configurations to better demonstrate the process. Some changes may need to be made to suit your needs.

Data Acquisition

“Data-Acquisition” is the process and the steps involved in extracting data from the source systems, (e.g., on-premise database).

We started off quickly analyzing some tools to understand the limitations and functionalities of data acquisition. We were hosted in Azure so we needed to adopt a solution that could be either Azure specific or generic across various cloud providers.

Debezium

We took a look at Debezium which is an open-source distributed platform for change data capture. It is a well-known solution in the open-source community providing the ability to capture the data changes and publish it into Kafka. It achieves this solution using Log capture thus avoiding any strain on the database.

Here is a wonderful blog on the advantages of adopting a Log-based capture vs. traditional data capture, via query.

Okay, so didn’t we use this approach? Well, as of this writing in March 2020:

The Oracle connector is incubating; meaning the published message format could change in future iterations.

The Oracle connector also requires a Golden Gate license, which means cost. We also currently did not have any Golden Gate implementations across our team.

Kafka JDBC Connector

Our choice was to use the de-facto Kafka JDBC source connector.

The Kafka JDBC connector offers a polling-based solution, whereby the database is queried at regular intervals. The data retrieved can be in bulk mode or incremental updates.

Connecting to the Database

While it sounds simple, you might face connection or configuration issues. Do not worry — here is a wonderful article that aids in solving some issues:

Blog: Kafka Connect Deep Dive — JDBC Source Connector

I recommend reading the section “Checking that the JDBC driver has been loaded” to confirm that the Oracle driver is loaded.

Confluent Kafka Connect Configuration

Here is the connect configuration which helps us to get a bulk data load of the table into Kafka:

name=orcl-source connector.class=io.confluent.connect.jdbc.JdbcSourceConnector tasks.max=1 connection.url=jdbc:oracle:thin:@localhost:49161:xe connection.user=books_admin connection.password=password dialect.name=OracleDatabaseDialect schema.pattern=APEX_040000 table.whitelist=TEST_DEMO_EMP # mode of operation mode=bulk #convertor definitions topic.prefix=tpc-avro- key.converter=org.apache.kafka.connect.json.JsonConverter header.converter=org.apache.kafka.connect.json.JsonConverter value.converter=io.confluent.connect.avro.AvroConverter value.converter.schema.registry.url=http://localhost:8081 #To handle data type conversion numeric.mapping=best_fit transforms=Cast transforms.Cast.type=org.apache.kafka.connect.transforms.Cast$Value transforms.Cast.spec=TESTSAL:int32

Data Type Conversion

When the data gets pulled and transformed into a Kafka message, you typically would not have any issue with a VARCHAR/String datatype in the source table. However, for numeric /timestamp data types you would need to handle some potential conversion.

Below is the connector configuration where we defined the value as JSON:

key.converter=org.apache.kafka.connect.json.JsonConverter value.converter=org.apache.kafka.connect.json.JsonConverter value.converter.schemas.enable=true

Below is the sample JSON message that gets published in the topic:

{ “schema”: { “type”: “struct”, “fields”: [ { “type”: “int32”,”optional”: true, “field”: “EMPNO”}, { “type”: “string”,”optional”: true,”field”: “ENAME”}, { “type”: “string”,”optional”: true,”field”: “JOB”}, { “type”: “int64”, “optional”: true, “field”: “MGR”}, { “type”: “int64”, “optional”: true, “name”: “org.apache.kafka.connect.data.Timestamp”,”version”: 1, “field”: “HIREDATE” }, { “type”: “double”, “optional”: true, “field”: “SAL” }, { “type”: “double”, “optional”: true, “field”: “COMM” }, { “type”: “int32”, “optional”: true, “field”: “DEPTNO”}, { “type”: “string”, “optional”: true, “field”: “WORK_CITY”}, { “type”: “string”, “optional”: true, “field”: “WORK_COUNTRY”}, { “type”: “bytes”, “optional”: true, “name”: “org.apache.kafka.connect.data.Decimal”, “version”: 1, “parameters”: { “scale”: “127”}, “field”: “TESTSAL” } ], “optional”: false, “name”: “TEST_DEMO_EMP” }, “payload”: { “EMPNO”: 10000, “ENAME”: “Anon-1k”, “JOB”: “Analyst”, “MGR”: 9047, “HIREDATE”: 1085270400000, “SAL”: 1300.0, “COMM”: null, “DEPTNO”: 40, “WORK_CITY”: “Tokyo”, “WORK_COUNTRY”: “Japan”, “TESTSAL”: “AeS7O40u/lY+QxzguxN0y/x/PcCXu7UV4y6aX2i/jS1riYg2bNzSAAAAAAAAAAAAAAAAAAAAAA==” } }

For the JSON value-based:

The message contains both the payload (data) and also the schema.

The schema element reflects the column data types.

The column ‘TESTSAL’; looks weird even though its a decimal why?

Hurdle: Handling Numeric Data Types

To start off, here is the table DDL:

CREATE TABLE “APEX_040000”.”TEST_DEMO_EMP” ( “EMPNO” NUMBER(8,0), “ENAME” VARCHAR2(30), “JOB” VARCHAR2(30), “MGR” NUMBER(14,0), “HIREDATE” DATE, “SAL” NUMBER(9,2), “COMM” NUMBER(9,2), “DEPTNO” NUMBER(5,0), “WORK_CITY” VARCHAR2(100), “WORK_COUNTRY” VARCHAR2(100), “TESTSAL” NUMBER );

When the numeric columns, e.g.: MGR, are defined with precision,e.g.: NUMBER(14,0), the converter is able to handle the conversion. So in the above JSON output, you would see the value “ MGR: 9047”.

When the numeric columns, e.g.: TESTSAL, are defined without precision, e.g.: NUMBER, the convertor converts this into a BYTE array and outputs this into the JSON. Hence the value is: “AeS7O40u/lY+QxzguxN0y/x/PcCXu7UV4y6aX2i/jS1riYg2bNzSAAAAAAAAAAAAAAAAAAAAAA==”

How this can be handled is explained in the section: Bytes, Decimals, Numerics of the previously mentioned blog post. I would recommend reading this section as well.

In our case, we do not have the option of updating the original table in the database. We did not like the approach of defining a query in the database, which had the appropriate precision definition, nor having the connector use this query.

We overcame this conversion by adopting the Single Message Transforms (SMT).

Single Message Transforms in Kafka Connect

Blog: How to Use Single Message Transforms in Kafka Connect

Below is the SMT defined in the configuration above:

transforms=Cast transforms.Cast.type=org.apache.kafka.connect.transforms.Cast$Value transforms.Cast.spec=TESTSAL:int32

Below is the sample JSON output, as you can see the conversion happens:

{“schema”:{“type”:”struct”,”fields”:[{“type”:”int32",”optional”:true,”field”:”EMPNO”},{“type”:”string”,”optional”:true,”field”:”ENAME”},{“type”:”string”,”optional”:true,”field”:”JOB”},{“type”:”int64",”optional”:true,”field”:”MGR”},{“type”:”int64",”optional”:true,”name”:”org.apache.kafka.connect.data.Timestamp”,”version”:1,”field”:”HIREDATE”},{“type”:”double”,”optional”:true,”field”:”SAL”},{“type”:”double”,”optional”:true,”field”:”COMM”},{“type”:”int32",”optional”:true,”field”:”DEPTNO”},{“type”:”string”,”optional”:true,”field”:”WORK_CITY”},{“type”:”string”,”optional”:true,”field”:”WORK_COUNTRY”},{“type”:”int32",”optional”:true,”field”:”TESTSAL”}],”optional”:false,”name”:”TEST_DEMO_EMP”},”payload”:{“EMPNO”:10000,”ENAME”:”Anon-1k”,”JOB”:”Analyst”,”MGR”:9047,”HIREDATE”:1085270400000,”SAL”:1300.0,”COMM”:null,”DEPTNO”:40,”WORK_CITY”:”Tokyo”,”WORK_COUNTRY”:”Japan”,”TESTSAL”:2100}}

Avro Convertors

It is generally recommended that the message be in Avro format. Below is the configuration:

key.converter=org.apache.kafka.connect.json.JsonConverter header.converter=org.apache.kafka.connect.json.JsonConverter value.converter=io.confluent.connect.avro.AvroConverter value.converter.schema.registry.url=http://localhost:8081

key.converter=org.apache.kafka.connect.json.JsonConverter header.converter=org.apache.kafka.connect.json.JsonConverter value.converter=io.confluent.connect.avro.AvroConverter value.converter.schema.registry.url=http://localhost:8081

Key & Header Convertor

The Avro converter does not work for Key & Header; they have to be a JSON convertor.

Handling Numeric data type

The configuration needs to have the SMT defined in case there is a number column in the table without any precision. If it’s not defined, an exception is below:
['Venkat Sekar']
['Data', 'Cloud Computing', 'Kafka', 'Snowflake', 'Open Source']