NiFi ExecuteSqlRecord не обрабатывает SQL запрос (Presto) - PullRequest
Новая
       2

NiFi ExecuteSqlRecord не обрабатывает SQL запрос (Presto)

1 голос
/ 29 февраля 2020

Я выполняю простой запрос Pivot в Presto через процессор «ExecuteSqlRecord», но в тот момент, когда я запускаю процессор, он выдает ошибку: Невозможно выполнить SQL выбрать .

Запрос: 

with map_date as 
(
 SELECT 
 vin, 
 epoch,
 timestamp,
 date,
 map_agg(signalName, value) as map_values
from hive.vehicle_signals.vehicle_signals_flat
 where date(date) = date('2020-02-28') 
 and vin = '000015'
 and signalName in ('timestamp','epoch','msgId','usec','vlan','vin','msgName','A','B','C','D','E','F','G','H',
 'J','K','L','M','N','O','P','R','S','T','U','V','W','X',
 'Y','Z','A1','A5','A2','A3','A4','A6','A8','A9','A10','A11','A12')
 GROUP BY vin, epoch, timestamp, date
order by timestamp asc 
)
SELECT
  epoch
, timestamp
, CASE WHEN element_at(map_values, 'A') IS NOT NULL THEN map_values['A'] ELSE NULL END AS A
, CASE WHEN element_at(map_values, 'B') IS NOT NULL THEN map_values['B'] ELSE NULL END AS B
, CASE WHEN element_at(map_values, 'C') IS NOT NULL THEN map_values['C'] ELSE NULL END AS C
, CASE WHEN element_at(map_values, 'D') IS NOT NULL THEN map_values['D'] ELSE NULL END AS D
, CASE WHEN element_at(map_values, 'E') IS NOT NULL THEN map_values['E'] ELSE NULL END AS E
, CASE WHEN element_at(map_values, 'F') IS NOT NULL THEN map_values['F'] ELSE NULL END AS F
, CASE WHEN element_at(map_values, 'G') IS NOT NULL THEN map_values['G'] ELSE NULL END AS G
, CASE WHEN element_at(map_values, 'H') IS NOT NULL THEN map_values['H'] ELSE NULL END AS H
, CASE WHEN element_at(map_values, 'J') IS NOT NULL THEN map_values['J'] ELSE NULL END AS J
, CASE WHEN element_at(map_values, 'K') IS NOT NULL THEN map_values['K'] ELSE NULL END AS K
, CASE WHEN element_at(map_values, 'L') IS NOT NULL THEN map_values['L'] ELSE NULL END AS L
, CASE WHEN element_at(map_values, 'M') IS NOT NULL THEN map_values['M'] ELSE NULL END AS M
, CASE WHEN element_at(map_values, 'N') IS NOT NULL THEN map_values['N'] ELSE NULL END AS N
, CASE WHEN element_at(map_values, 'O') IS NOT NULL THEN map_values['O'] ELSE NULL END AS O
, CASE WHEN element_at(map_values, 'P') IS NOT NULL THEN map_values['P'] ELSE NULL END AS P
, CASE WHEN element_at(map_values, 'R') IS NOT NULL THEN map_values['R'] ELSE NULL END AS R
, CASE WHEN element_at(map_values, 'S') IS NOT NULL THEN map_values['S'] ELSE NULL END AS S
, CASE WHEN element_at(map_values, 'T') IS NOT NULL THEN map_values['T'] ELSE NULL END AS T
, CASE WHEN element_at(map_values, 'U') IS NOT NULL THEN map_values['U'] ELSE NULL END AS U
, CASE WHEN element_at(map_values, 'V') IS NOT NULL THEN map_values['V'] ELSE NULL END AS V
, CASE WHEN element_at(map_values, 'W') IS NOT NULL THEN map_values['W'] ELSE NULL END AS W
, CASE WHEN element_at(map_values, 'X') IS NOT NULL THEN map_values['X'] ELSE NULL END AS X
, CASE WHEN element_at(map_values, 'Y') IS NOT NULL THEN map_values['Y'] ELSE NULL END AS Y
, CASE WHEN element_at(map_values, 'Z') IS NOT NULL THEN map_values['Z'] ELSE NULL END AS Z
, CASE WHEN element_at(map_values, 'A1') IS NOT NULL THEN map_values['A1'] ELSE NULL END AS A1
, CASE WHEN element_at(map_values, 'A2') IS NOT NULL THEN map_values['A2'] ELSE NULL END AS A2
, CASE WHEN element_at(map_values, 'A3') IS NOT NULL THEN map_values['A3'] ELSE NULL END AS A3
, CASE WHEN element_at(map_values, 'A4') IS NOT NULL THEN map_values['A4'] ELSE NULL END AS A4
, CASE WHEN element_at(map_values, 'A5') IS NOT NULL THEN map_values['A5'] ELSE NULL END AS A5
, CASE WHEN element_at(map_values, 'A6') IS NOT NULL THEN map_values['A6'] ELSE NULL END AS A6
, CASE WHEN element_at(map_values, 'A8') IS NOT NULL THEN map_values['A8'] ELSE NULL END AS A8
, CASE WHEN element_at(map_values, 'A9') IS NOT NULL THEN map_values['A9'] ELSE NULL END AS A9
, CASE WHEN element_at(map_values, 'A10') IS NOT NULL THEN map_values['A10'] ELSE NULL END AS A10
, CASE WHEN element_at(map_values, 'A11') IS NOT NULL THEN map_values['A11'] ELSE NULL END AS A11
, CASE WHEN element_at(map_values, 'A12') IS NOT NULL THEN map_values['A12'] ELSE NULL END AS A12
, vin
, date
from map_date

Я могу отлично выполнить запрос ABOVE в ExecuteSqlRecord, если я поворачиваю только ~ 20 полей, но при повороте ~ 37 полей тогда процессор ExecuteSqlRecord немедленно завершается ошибкой.

Интересно, что я могу успешно выполнить один и тот же сводный запрос с ~ 37 полями в Presto CLI и QueryDatabaseTable! Пожалуйста, помогите мне добиться этого в процессоре ExecuteSQLRecord, поскольку он имеет дополнительное преимущество записи выходного файла потока в любом желаемом формате (для моего случая я пишу вывод запроса SQL как CSV)

Здесь полная трассировка стека ошибки, 

 java.sql.SQLException: Error executing query. No FlowFile to route to failure: java.sql.SQLException: Error executing query

java.sql.SQLException: Error executing query

 at com.facebook.presto.jdbc.PrestoStatement.internalExecute(PrestoStatement.java:279)

 at com.facebook.presto.jdbc.PrestoStatement.execute(PrestoStatement.java:228)

 at com.facebook.presto.jdbc.PrestoPreparedStatement.<init>(PrestoPreparedStatement.java:80)

 at com.facebook.presto.jdbc.PrestoConnection.prepareStatement(PrestoConnection.java:129)

 at org.apache.commons.dbcp2.DelegatingConnection.prepareStatement(DelegatingConnection.java:303)

 at org.apache.commons.dbcp2.DelegatingConnection.prepareStatement(DelegatingConnection.java:303)

 at org.apache.nifi.processors.standard.AbstractExecuteSQL.onTrigger(AbstractExecuteSQL.java:237)

 at org.apache.nifi.processor.AbstractProcessor.onTrigger(AbstractProcessor.java:27)

 at org.apache.nifi.controller.StandardProcessorNode.onTrigger(StandardProcessorNode.java:1176)

 at org.apache.nifi.controller.tasks.ConnectableTask.invoke(ConnectableTask.java:213)

 at org.apache.nifi.controller.scheduling.TimerDrivenSchedulingAgent$1.run(TimerDrivenSchedulingAgent.java:117)

 at org.apache.nifi.engine.FlowEngine$2.run(FlowEngine.java:110)

 at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)

 at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)

 at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)

 at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)

 at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)

 at java.base/java.lang.Thread.run(Thread.java:834)

Caused by: java.lang.RuntimeException: Error fetching next at http://presto.testk8s.content.mywebspace.com:8889/b7/journal/20245300229_0354554_08r940_as8y4/2?slug=78234778257265 returned an invalid response: JsonResponse{statusCode=500, statusMessage=Server Error, headers={connection=[close]}, hasValue=false} [Error: ]

 at com.facebook.presto.jdbc.internal.client.StatementClientV1.requestFailedException(StatementClientV1.java:446)

 at com.facebook.presto.jdbc.internal.client.StatementClientV1.advance(StatementClientV1.java:386)

 at com.facebook.presto.jdbc.PrestoResultSet.getColumns(PrestoResultSet.java:1742)

 at com.facebook.presto.jdbc.PrestoResultSet.<init>(PrestoResultSet.java:119)

 at com.facebook.presto.jdbc.PrestoStatement.internalExecute(PrestoStatement.java:250)

 ... 17 common frames omitted
...