1 从Kafka接收消息，存储到数据库中。
（1） ConsumerKafka processor
（2）Execute Scripts Processor
我这里是使用JS脚本进行处理。 还有很多其他语言的脚本。
var flowFile = session.get();
if (flowFile != null) {var IOUtils = Java.type("org.apache.commons.io.IOUtils");var StreamCallback = Java.type("org.apache.nifi.processor.io.StreamCallback");var StandardCharsets = Java.type("java.nio.charset.StandardCharsets");var DateFormatUtils=Java.type("org.apache.commons.lang3.time.DateFormatUtils");// var dataType=flowFile.getAttribute('data_type')// var FLAG=flowFile.getAttribute('flag')var tm = null;try {flowFile = session.write(flowFile, new StreamCallback(function (inputStream, outputStream) {var inputText = IOUtils.toString(inputStream, StandardCharsets.UTF_8);var msg = JSON.parse(inputText);var stationId = msg['stationId'];var stationName = msg['stationName'];var deviceId = msg['deviceId'];var deviceName = msg['deviceName'];var deviceNo = msg['deviceNo'];var receiveType = msg['receiveType'];var createAt = msg['createAt'];var createAtString=DateFormatUtils.format(Number(createAt),'yyyy-MM-dd HH:mm:ss');var obTime = msg['obTime'];var obDate = msg['obDate'];var obDateString=DateFormatUtils.format(Number(obDate),'yyyy-MM-dd HH:mm:ss');var order = msg['order'];var distance = msg['distance'];var channel1SignalStrength = msg['channel1SignalStrength']var powerVoltage = msg['powerVoltage']var sql = 'insert into "SJZT_ODS"."water_data_distance"('+ '"station_id", "station_name", "device_id", "device_name", "device_no", "receive_type", "create_at", "ob_time", "ob_date", "order", "distance", "channel1_signal_strength", "power_voltage")'+ 'VALUES('+ stationId + ', \'' + stationName + '\', ' + deviceId + ', \'' + deviceName + '\', \'' + deviceNo + '\', ' + receiveType + ', \'' + createAtString + '\', \'' + obTime + '\', \'' + obDateString + '\', ' + order + ', ' + distance + ', ' + channel1SignalStrength + ', ' + powerVoltage + ')';outputStream.write(sql.getBytes(StandardCharsets.UTF_8));}));// flowFile = session.putAttribute(flowFile, "tm",tableName);session.transfer(flowFile, REL_SUCCESS);} catch (e) {flowFile = session.putAttribute(flowFile, "rsvr.transfer.error", e);session.transfer(flowFile, REL_FAILURE);}
}
注意： 这里只是生成了一个sql字符串，并没有执行sql，因此需要后面的processor来执行sql语句。
（3）PutSql processor
注意：autocommit要设置为true，否则看不到数据库里面的数据的。
2 将一堆Processors移动到一个Group里面界面操作
貌似没有直接的移动操作。
（1） Ctrl + A 全选要移动的processors
（2） 点击左边的group按钮
（3）为新的Group命名
（4）好了。选中的所有的processors都移动到了自己新创建的group中了。
参考材料
[1] https://blog.csdn.net/guijianchouxyz/article/details/120340154
