Groovy скрипт не работает в процессоре исполняемого скрипта nifi - PullRequest
Новая
       7

Groovy скрипт не работает в процессоре исполняемого скрипта nifi

0 голосов
/ 18 мая 2018

Я пытаюсь выполнить что-то через процессор исполняемого сценария;Отличный код внутри.В коде я пытаюсь создать скрипт scala, который должен быть запущен на спарк в другом процессоре.

// Get flow file
def flowFile = session.get()
if (!flowFile) return
// Create output directory
def userInputDir = flowFile.getAttribute("user.input.path")
def finalFolder = new File(userInputDir + "/" + innerDir)

try
{
if (!finalFolder.exists()) finalFolder.mkdirs()
// Write script
file = "spark.sqlContext.setConf(\"hive.exec.dynamic.partition\", \"true\")\n"
file = file + "spark.sqlContext.setConf(\"hive.exec.dynamic.partition.mode\", \"nonstrict\")\n"
file = file + "import org.apache.spark.sql._"
file = file + "\n"
file = file + "import java.io._"
file = file + "\n"

}.,,Остальные остальные шаги - добавление некоторых других специфических для искры команд в переменную script.Скрипт огромен, поэтому пропускаем полную вставку кода.Наконец, закрытие с перехватом 

// Output file path
flowFile = session.putAttribute(flowFile, "generatedScript", scalaFile.getCanonicalPath())
session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e)
{
 log.info("File: {}\n", finalFolder.file)
 session.transfer(flowFile, REL_FAILURE)
}

Процессор даже не начинает запускать скрипт groovy и завершается с ошибкой: 

groovy.lang.MissingPropertyException: No such property: script for calss: javal.io.File

С помощью оператора 'даже не начинаетсяto start 'означает, что предыдущая очередь не пуста, и процессор выдает ошибку.Я предполагаю, что это проблема синтаксиса, но я не нахожу никаких синтаксических проблем, связанных с сценарием.Я также попытался запустить скрипт в оболочке groovy на локальной машине, и там тоже произошла та же ошибка, но не возникла проблема с синтаксисом.

При поиске ошибки я получил предложение включить импорт в сценарий, но даже после включения соответствующего импорта ошибка та же.

Есть какие-нибудь подсказки?

1 Ответ

0 голосов
/ 22 мая 2018

Вы ссылаетесь на переменную "innerDir", которая нигде не определена.Вы ожидаете, что пользователь добавит пользовательское свойство в ExecuteScript с именем innerDir?Если это так, переменная innerDir в сценарии является объектом PropertyValue, поэтому вам нужно вызвать getValue () для него, чтобы получить фактическое значение свойства: 

innerDir.value

Также вы ссылаетесь на scalaFile.getCanonicalPath (), но scalaFile не определен выше, и getCanonicalPath () не даст вам содержимое скрипта, это то, что вы имели в виду?

Я переделал приведенный выше частичный скрипт, чтобы предположить, что innerDir является пользователем-определенное свойство, и вы записываете содержимое переменной файла в файл, на который указывает scalaFile;также я сделал это более Groovy, используя heredoc вместо добавления к переменной файла: 

// Get flow file
def flowFile = session.get()
if (!flowFile) return
// Create output directory
def userInputDir = flowFile.getAttribute("user.input.path")
def finalFolder = new File(userInputDir + "/" + innerDir?.value ?: '')

try
{
  if (!finalFolder.exists()) finalFolder.mkdirs()
  // Write script
  file = 
"""
  spark.sqlContext.setConf("hive.exec.dynamic.partition", "true")
  spark.sqlContext.setConf("hive.exec.dynamic.partition.mode", "nonstrict")
  import org.apache.spark.sql._
  import java.io._
"""
  scalaFile = new File(finalFolder, 'script.scala')
  scalaFile.withWriter {w -> w.write(file)}
  // Output file path
  flowFile = session.putAttribute(flowFile, "generatedScript", scalaFile.canonicalPath)
  session.transfer(flowFile, REL_SUCCESS)
}
catch(Exception e) {
  log.info("File: {}\n", finalFolder.file)
  session.transfer(flowFile, REL_FAILURE)
}
...