Day15 NiFi - 与 RDB 对接设定

今天来介绍一下 NiFi 如何对 Relation Database 做整合操作，这边我会以 MySQL 来做一个介绍。

Build MySQL Container

首先我们一样透过 docker 来建立 mysql，这边一样给读者们一个 docker-compose.yaml sample:

version: '3'

services:
    nifi:
        image: nifi-sample
        container_name: nifi-service
        restart: always
        ports:
            - 8443:8443/tcp
            - 8080:8080/tcp
        env_file: .env
        environment:
            SINGLE_USER_CREDENTIALS_USERNAME: ${NIFI_USERNAME}
            SINGLE_USER_CREDENTIALS_PASSWORD: ${NIFI_PASSWORD}
            AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
            AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
            AWS_REGION: ${AWS_REGION}
        networks:
            - nifi-network

    nifi-registry:
        image: apache/nifi-registry:1.14.0
        container_name: nifi-registry-service
        restart: always
        ports:
            - 18080:18080/tcp
        networks:
            - nifi-network
    
    mysql:
        image: mysql:5.7
        container_name: nifi-rdb
        restart: always
        ports:
            - 3306:3306/tcp
        env_file: .env
        environment:
            MYSQL_ROOT_PASSWORD: ${DB_ADMIN_PASSWORD}
            MYSQL_DATABASE: ${DB_NAME}
        networks:
            - nifi-network
networks:
    nifi-network:

接着透过 docker-compose up -d 建立起来就没有问题了。

MySQL JDBC

至於如何在 NiFi 中来对 MySQL 做对接了，其实NiFi 主要是提通的 General Processor 和 Controller Service Settings，也就是说对於外部的 RDB 或 Query Engine等服务，所采用的设定都是一样的，唯一不一样的是 JDBC Driver。

因为 NiFi 本身是用 JAVA 所开发出来的，所以对於外部的资料库时必须要透过 JDBC 的 Driver 才可以做资料对接，这是相对比较有一点小麻烦，因为我们必须先下载 JDBC Driver 到指定的 Container 的路径下。

这边给读者们 MySQL JDBC Driver 下载的 link，请点此，再依据自己的 MySQL 版本下载对应的 JDBC，Driver name 是 com.mysql.jdbc.Driver。

How to use?

接着我们到 Controller Service 来选择 DBConnectionPool 来做对接的设定:

接着我们设定好其中必要的 Properties:

• Database Connection URL:
  用来指定 JDBC URL，例如 jdbc:mysql://172.17.0.1:3306/nifi?user=[username]&password=[password]
• Database Driver Class Name:
  用来指定 JDBC Class Name，例如 com.mysql.jdbc.Driver
• Database Driver Location:
  用来指定 JDBC Driver JAR 档的路径，例如/tmp/mysql-connector-java-5.1.49.jar

接下来我们就可以来使用 Processor 对 MySQL 做整合了，今天会用 PutSQL 和 QueryDatabaseTable 来做介绍：

PutSQL

这个 Processor 主要是让我们可以对 Database 做 SQL 的操作，ex. INSERT, UPDATE, DELETE 等，所以他在设定上也是相对简单，参考如下：

• JDBC Connection Pool:
  指定好我们刚刚设定的 DBConnectionPool 的 Controller Service。
• SQL Statement:
  这边可以输入 SQL 的操作，这边以 INSERT 为例。

透过以上建制之後，我们就可以对 MySQL 做资料的操作了。

QueryDatabaseTable

这个 Processor 主要是用来从 DB 到读取资料出来用的，但就无法做 Insert, update 等操作，我们来看一下他的设定：

• Database Connectoin Pooling Service:
  一样指定刚刚设定好的 Controller Service。
• Database Type:
  这边就是选择 DB 类型，我们可以选择 MySQL。此外他也有支援 Orable, MSSQL 等类型
• Table Name:
  输入你要拉资料的 table 名称
• Columns to Return
  回传你要的 Column name，并以 , 做分隔
• Additional WHERE clause
  要加入的条件筛选。

简单来说，你可以想像成我们就常用的 SELECT..FROM..WHERE.. 的语法拆解到这个 Processor 的各个设定上。当然你也可以透过 PutSQL 做到一样的效果。

小总结

透过今天简单的小分享，想必大家应该知道如何从 NiFi 的角度出发来向 RDB table 的资料做操作，虽然今天是以 MySQL 为例，但其实其他的 RDB 也都是一样的操作，差别在於 JDBC 那部份的设定而已，读者们可以再去尝试看看喔。

既然已经讲完了关连式资料库的整合，明天就会来讲如何与 MongoDB 做整合，这部分就不用那麽麻烦了，因为 NiFi 就有内建好的 Processor 可以做使用，所以读者们也不必太担心。

Reference

• Aapche NiFi Document