如何通过NiFi将binary.zip上传至SQL Server varbinary字段

目标表结构 #

目标表结构定义如下：

CREATE TABLE myTable ( zipFile varbinary(MAX) );

初始NiFi流程配置 #

初始搭建的处理链路逻辑简单，各处理器配置如下：

-> GetFile: 
         filter:binary.zip

  -> UpdateAttribute:
         sql.args.1.type  = -3    # 按JDBC类型枚举对应varbinary类型
         sql.args.1.value =  ???  # 尝试多种取值均未生效，不确定正确填入内容
         sql.args.1.format=  ???  # 不确定是否为必填项，曾尝试配置为'hex'

   -> PutSQL:
         SQLstatement= INSERT INTO myTable (zip_file) VALUES (?);

核心疑问 #

配置过程中最核心的问题是sql.args.1.value项的正确取值规则不明确，按业务逻辑该位置应传入FlowFile的内容载荷，但多次尝试不同取值后，PutSQL处理器的INSERT执行逻辑始终无法正常生效。

可行实现方案 #

PutSQL处理器对二进制流类型的参数传递存在已知限制，可通过ExecuteScript处理器运行Groovy脚本绕过该限制实现二进制文件写入，方案支持通过FlowFile属性传递自定义配置参数，无需硬编码连接、表名信息，脚本内容如下：

import java.nio.charset.StandardCharsets
import org.apache.nifi.controller.ControllerService
import groovy.sql.Sql

def flowFile = session.get()
def lookup = context.controllerServiceLookup
def dbServiceName = flowFile.getAttribute('DatabaseConnectionPoolName')  
def tableName = flowFile.getAttribute('table_name')
def fieldName = flowFile.getAttribute('field_name')

def dbcpServiceId = lookup.getControllerServiceIdentifiers(ControllerService).find
{ cs -> lookup.getControllerServiceName(cs) == dbServiceName }

def conn = lookup.getControllerService(dbcpServiceId)?.getConnection()
def sql = new Sql(conn)


flowFile.read{ rawIn->
    def parms = [rawIn ]
    sql.executeInsert "INSERT INTO " + tableName + " (date, "+ fieldName + ")  VALUES (CAST( GETDATE() AS Date ) , ?) ", parms
}
conn?.close()

if(!flowFile) return
session.transfer(flowFile, REL_SUCCESS)
session.commit()

使用注意：运行脚本前需提前给FlowFile设置三个属性：DatabaseConnectionPoolName填写NiFi中已配置好的数据库连接池服务名，table_name填写目标写入表名，field_name填写存储二进制内容的varbinary列名，脚本会自动读取对应属性完成写入操作。

内容的提问来源于stack exchange，提问作者Leonardo