Running Flink SQL Against Kafka Using a Schema Registry Catalog
Published: 2020-10-22 13:55:54

---
title: Running Flink SQL Against Kafka Using a Schema Registry Catalog
published: true
date: 2020-10-21 17:18:00 UTC
tags: 
canonical_url: 
---

# [FLaNK]:  Running Apache Flink SQL Against Kafka Using a Schema Registry Catalog

  

  

There are a few things you can do when you are sending data from Apache NiFi to Apache Kafka to maximize it's availability to Flink SQL queries through the catalogs.

  

  

### AvroWriter

[![](https://1.bp.blogspot.com/--8XZOovAidY/X5Bs2Q6yZKI/AAAAAAAAbxE/w3kULdgmDvkK6wGt4Yry2NS6KQA_DKtTACLcBGAsYHQ/w644-h356/avrowriter.png)](https://1.bp.blogspot.com/--8XZOovAidY/X5Bs2Q6yZKI/AAAAAAAAbxE/w3kULdgmDvkK6wGt4Yry2NS6KQA_DKtTACLcBGAsYHQ/s1205/avrowriter.png)
  

  

### JSONReader

[![](https://1.bp.blogspot.com/-rS_8xtM9O7w/X5Bs2dsO8DI/AAAAAAAAbxM/k3IOv4vdWKQMG-jEuNvQpCNb9y2OkLIlwCLcBGAsYHQ/w619-h377/jsonreader.png)](https://1.bp.blogspot.com/-rS_8xtM9O7w/X5Bs2dsO8DI/AAAAAAAAbxM/k3IOv4vdWKQMG-jEuNvQpCNb9y2OkLIlwCLcBGAsYHQ/s957/jsonreader.png)

  

  

  

### **Producing Kafka Messages**

  

Make sure you set **AvroRecordSetWriter** and set a **Message Key Field.**

  

  

[![](https://1.bp.blogspot.com/-W3qw6E5QP3c/X5Bs2YYxWnI/AAAAAAAAbxI/3g4f_W_zk_kRr_a77KdI8E49--gabQ1cgCLcBGAsYHQ/w580-h542/publishkafka1.png)](https://1.bp.blogspot.com/-W3qw6E5QP3c/X5Bs2YYxWnI/AAAAAAAAbxI/3g4f_W_zk_kRr_a77KdI8E49--gabQ1cgCLcBGAsYHQ/s1054/publishkafka1.png)
  

[![](https://1.bp.blogspot.com/-ZG5C0o-Rli0/X5Bs2twhkDI/AAAAAAAAbxQ/BEFZy0MhHsQvmbStg_f_Fy3G_KgfwW72gCLcBGAsYHQ/w558-h438/publishkafka2.png)](https://1.bp.blogspot.com/-ZG5C0o-Rli0/X5Bs2twhkDI/AAAAAAAAbxQ/BEFZy0MhHsQvmbStg_f_Fy3G_KgfwW72gCLcBGAsYHQ/s860/publishkafka2.png)

  

  

A great way to work with Flink SQL is to connect to the Cloudera Schema Registry.   It let's you define your schema once them use it in Apache NiFi, Apache Kafka Connect, Apache Spark, Java Microservices 

#### Setup

  

See:   [https://docs.cloudera.com/csa/1.2.0/sql-client/topics/csa-sql-client-config.html](https://docs.cloudera.com/csa/1.2.0/sql-client/topics/csa-sql-client-config.html)

See:   [https://docs.cloudera.com/csa/1.2.0/installation/topics/csa-hdfs-home-install.html](https://docs.cloudera.com/csa/1.2.0/installation/topics/csa-hdfs-home-install.html)

  

Make sure you setup your HDFS directory for use by Flink which keeps history and other important information in HDFS.

#### 

HADOOP\_USER\_NAME=hdfs hdfs dfs -mkdir /user/root

HADOOP\_USER\_NAME=hdfs hdfs dfs -chown root:root /user/root

  

#### SQL-ENV.YAML:

| configuration: |
| | execution.target: yarn-session |
| | |
| | catalogs: |
| | - name: registry |
| | type: cloudera-registry |
| | # Registry Client standard properties |
| | registry.properties.schema.registry.url: http://edge2ai-1.dim.local:7788/api/v1 |
| | # registry.properties.key: |
| | # Registry Client SSL properties |
| | # Kafka Connector properties |
| | connector.properties.bootstrap.servers: edge2ai-1.dim.local:9092 |
| | connector.startup-mode: earliest-offset |
| | - name: kudu |
| | type: kudu |
| | kudu.masters: edge2ai-1.dim.local:7051 |

#### CLI:

flink-sql-client embedded -e sql-env.yaml

#### We now have access to Kudu and Schema Registry catalogs of tables.   This let's use start querying, joining and filtering any of these multiple tables without having to recreate or redefine them.

  

SELECT \* FROM events

  

#### Code:

- [https://github.com/tspannhw/FlinkSQLWithCatalogsDemo](https://github.com/tspannhw/FlinkSQLWithCatalogsDemo)
- [https://github.com/tspannhw/ApacheConAtHome2020/tree/main/schemas](https://github.com/tspannhw/ApacheConAtHome2020/tree/main/schemas)
- [https://docs.cloudera.com/csa/1.2.0/overview/topics/csa-overview.html](https://docs.cloudera.com/csa/1.2.0/overview/topics/csa-overview.html)
- [https://docs.cloudera.com/csa/1.2.0/flink-sql-table-api/topics/csa-schemaregistry-catalog.html](https://docs.cloudera.com/csa/1.2.0/flink-sql-table-api/topics/csa-schemaregistry-catalog.html)
- [https://github.com/tspannhw/FlinkForwardGlobal2020](https://github.com/tspannhw/FlinkForwardGlobal2020)

[![](https://docs.cloudera.com/csa/1.2.0/overview/images/csa-architecture.png)](https://docs.cloudera.com/csa/1.2.0/overview/images/csa-architecture.png)