使用 Groovy Nifi 读取流文件答案

【问题标题】：Read FlowFile with Groovy Nifi使用 Groovy Nifi 读取流文件
【发布时间】：2018-03-06 02:10:23
【问题描述】：

我尝试将 Avro 文件转换为 SQL 请求。我的文件是这样的：

{
  "type" : "record",
  "name" : "warranty",
  "doc" : "Schema generated by Kite",
  "fields" : [ {
    "name" : "id",
    "type" : "long",
    "doc" : "Type inferred from '1'"
  }, {
    "name" : "train_id",
    "type" : "long",
    "doc" : "Type inferred from '21691'"
  }, {
    "name" : "siemens_nr",
    "type" : "string",
    "doc" : "Type inferred from 'Loco-001'"
  }, {
    "name" : "uic_nr",
    "type" : "long",
    "doc" : "Type inferred from '193901'"
  }, {
    "name" : "Configuration",
    "type" : "string",
    "doc" : "Type inferred from 'ZP28'"
  }, {
    "name" : "Warranty_Status",
    "type" : "string",
    "doc" : "Type inferred from 'Out_of_Warranty'"
  }, {
    "name" : "Warranty_Data_Type",
    "type" : "string",
    "doc" : "Type inferred from 'Real_based_on_preliminary_acceptance_date'"
  }

我的代码是：

import groovy.json.JsonSlurper

def ff = session.get()
if(!ff)return

//parse afro schema from flow file content
def schema = ff.read().withReader("UTF-8"){ new JsonSlurper().parse(it) }

//define type mapping
def typeMap = [
    "string"            : "varchar(255)",
    "long"              : "numeric(10)",
    [ "null", "string" ]: "varchar(255)",
    [ "null", "long" ]  : "numeric(10)",
]
//build create table statement
def createTable = "create table ${schema.name} (" +
    schema.fields.collect{ "\n  ${it.name.padRight(39)} ${typeMap[it.type]}" }.join(',') +
    "\n)"

//execute statement through the custom defined property
//SQL.mydb references http://docs.groovy-lang.org/2.4.10/html/api/groovy/sql/Sql.html object
SQL.mydb.execute(createTable)

//transfer flow file to success
REL_SUCCESS << ff

我得到了这个错误：

ERROR nifi.processors.script.ExecuteScript ExecuteScript[id=e65b733e-0161-1000-45f0-3264d6fb51dd] ExecuteSc$ Possible solutions: getId(), find(), grep(), each(groovy.lang.Closure), find(groovy.lang.Closure), grep(java.lang.Object); rolling back session: {} org.apache.nifi.processor.exception.ProcessException: javax.script.ScriptException: javax.script.ScriptException: groovy.lang.MissingMethodException: No signature of m$ Possible solutions: getId(), find(), grep(), each(groovy.lang.Closure), find(groovy.lang.Closure), grep(java.lang.Object)

谁能帮帮我

【问题讨论】：

标签： json groovy apache-nifi

【解决方案1】：

这引用了来自another SO post 的脚本，我在那里发表了评论，并在different forum 上提供了答案，为了完整起见，我将在此处复制：

变量 createTable 是 GString，而不是 Java String。这会导致调用 Sql.execute(GString)，它将嵌入的表达式转换为参数，并且您不能将参数用于表名。请改用以下内容：

SQL.mydb.execute(createTable.toString())

这会导致Sql.execute(String)的调用，它不会尝试参数化语句。

【讨论】：