We have seven different branches/lines of flow and at the terminal processors, we want to merge the output FlowFiles. These seven branches starts from a singe GenerateFlowFile, which in turn is triggered to run 3 times a day (1 AM, 7 AM and 12 PM). However, the execution time of the branches are different, so we need for each one of them to finish. At the merging part (funnel + MergeContent processor), we use the batch-time (trigger timestamp) as an attribute in order to bundle the 7 output FlowFiles. Things work out well for the most part, until one time, one of the branches has some mistake that it outputted two FlowFiles for same cycle instead of one. So for example, these FlowFiles were bundled together, while Branch 7 is not yet done: FlowFile1 FlowFile2 FlowFile3 FlowFile4 FlowFile5 FlowFile6 FlowFile6 Assuming we use a FlowFile attribute identifying the branch name, how can we bundle the same batch-time FlowFiles while guaranteeing that all branches are covered in the bundle? ... View more

I am prototyping a flow where a JSON will be split to fragments then after some intermediate processing, will be routed to Wait and Notify processors (basically they diverge after some time after some number of processes). On the side of Wait, I want all fragments to be released almost simultaneously once Notify that the last fragment ( max(fragment.index) = fragment.count - 1 ) have arrived. My assumption is that fragment.index follows the processing sequence right after the SplitJSON (is this correct?). If the previous assumption is correct, let us assume that the intermediate processes after SplitJSON up to Notify preserves such ordering. How do I configure Wait and Notify for it to work this way? Is there a better solution than what I am thinking? ... View more

I have an InvokeHTTP processor and the REST endpoint it calls requires PATCH method. However, when we send the FlowFile (we get the authentication part OK), we are getting: method PATCH must have a request body For the InvokeHTTP processor, I'm pretty sure that the we configured the Send Message Body to true (which is the default) and Content-Type as application/json. We enabled the endpoint to accept POST request, for the sake of testing and we've got the JSON payload accepted. So I wonder if there is an issue if we use PATCH. ... View more

According to JoltTransformJSON documentation, the JoltSpecification properties supports Expression Language. What I want is to use the FlowFile attribute containing the Jolt spec. Unfortunately, this is invalid as signified by the caution symbol. Is using attribute for Jolt specification property allowed? ... View more

Is it possible to place a NiFi processors name or id to a flowfile that it outputs? For example a Processor InvokeHTTP I named as CallRESTEndpoint, the flowfile that would come out of it will have attribute, like origin:CallRESTEndpoint? Or if possible to create a chain of origin that will append the processor's name or id to this attribute. I need this in error / failure handling. If you can share a better idea in tracing errors / failure, that'd be a good idea. (But please answer the original question also, if there is one.) ... View more

We have a custom query that we wish to put inside the ExecuteSQL processor, but this one proves a bit tricky: SELECT * FROM mytable AS OF TIMESTAMP TO_TIMESTAMP('${now():format('yyyy-MM-dd HH:mm:ss')}', 'YYYY-MM-DD HH24:MI:SS'); The processor is CRON driven where we run it every 15 minutes. The expression language inside the first single quotes is NiFi's expression language to get the current time in the format specified. Note that the expression language: ${now():format('yyyy-MM-dd HH:mm:ss')} itself uses single quotes. I tried substituting double quotes for the Oracle query, but Oracle won't accept the syntax as valid. So how do we substitute the intended date-time string inside the first single-quotes? ... View more

We have an SFDC REST endpoint (let's call this UpdateSFDC) where we stream and post database changes, in JSON format, to reflect it to SalesForce object store. Before we get to that endpoint, we need to Login to SalesForce first to this endpoint: https://test.salesforce.com/services/oauth2/token So we use InvokeHTTP (let's call this Login2SFDC) and appends the parameters to the Remote URL field in the processor: https://test.salesforce.com/services/oauth2/token?grant_type=password&client_id=<client_id>&client_secret=<client_secret>&username=<username>&password=<password + key> This will respond with a token which we need to extract and add as an attribute in every JSON format flowfile (let's call this CreateSignedRequest). Here's a NiFi schematic: The Login2SFDC is configure this way: However, this is quite inefficient because for every JSON flowfile, we need to login to SalesForce everytime, which will add overhead, not to mention the risk that Login2SFDC REST endpoint will eventually throw some sort of a max limit request reached. Rather, we want to login once and as long as the token is valid (not yet expired), then we will just add the token as an attribute to the JSON flowfile and go to UpdateSFDC REST endpoint straight (via CreateSignedRequest). So I will convert the Login2SFDC processor from InvokeHTTP to ExecuteScript. I have 3 questions for this problem: 1. How do I this in ECMAScript and/or Python? When I use their respective native HTTP client functionalities, they do not seem to work. Should I be importing and using Java libraries (like HttpClient) instead in which case, the scripting language is simply a wrapper to Java libraries? 2. I am successful in getting a response from Login2SFDC REST endpoint when in NiFi. I tried writing a corresponding Java code (using same exact endpoint and credentials) using HttpClient library: CloseableHttpClient client = HttpClients.createDefault(); String urlStr = "https://test.salesforce.com/services/oauth2/token?grant_type=password&client_id=<client_id>&client_secret=<client_secret>&username=<username>&password=<password + key>"; HttpPost httpPost = new HttpPost(urlStr); String jsonPayload = "{\"qty\":100,\"name\":\"iPad 4\"}"; StringEntity entity = new StringEntity(jsonPayload); httpPost.setEntity(entity); httpPost.setHeader("Accept", "application/json"); httpPost.setHeader("Content-type", "application/json"); CloseableHttpResponse response = client.execute(httpPost); System.out.println(response.getStatusLine().getStatusCode()); InputStream is = response.getEntity().getContent(); String responseStr = IOUtils.toString(is, "UTF-8"); System.out.println(responseStr); client.close(); but I get this response: Status Code 400 {"error":"invalid_grant","error_description":"authentication failure"} Am I missing something in the Java code? 3. Is there a better NiFi flow design for this kind of problem? Hi @Matt Burgess, can you help? ... View more

We have two tables in Hive, the first one, let's call it Table A, is where the raw data is being dumped. The data in Table B is derived from Table A after some HiveQL transformation. (I understand there are no stored procedures in Hive). When new data will be inserted on Table A, how do we automatically trigger the process of doing the same transformation and appending the result on Table B? ... View more

I'm following a tutorial here from MapR in making a basic Kafka producer/consumer: https://github.com/mapr-demos/kafka-sample-programs I tried it both on HDP Sandbox VM 2.4 (Kafka v0.9.2) and HDP 2.5 on single-node Azure VM (Kafka v0.10.0.2). Here is my Producer code: package com.whatever.kafka.basic; ... import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.ProducerConfig; public class BasicProducer { public static void main(String[] args) throws IOException { /* comment-out snippet from MapR example; hard-code properties instead KafkaProducer<String, String> producer; try (InputStream props = Resources.getResource("producer.props").openStream()) { Properties properties = new Properties(); properties.load(props); producer = new KafkaProducer<String, String>(properties); } */ Properties props = new Properties(); // props.put("bootstrap.servers", "<azure-vm-fqdn>:6667"); // for HDP 2.5 Azure VM props.put("bootstrap.servers", "sandbox.hortonworks.com:6667"); // for HDP 2.4 Sandbox VM props.put("acks", "all"); props.put("retries", 0); props.put("batch.size", 16384); props.put("auto.commit.interval.ms", 1000); props.put("linger.ms", 0); props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); props.put("block.on.buffer.full", true); final KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props); try { for (int i = 0; i < 1000000; i++) { System.out.println("Test 1"); // prints to console and stops on the next line producer.send(new ProducerRecord<String, String>( "fast-messages", String.format("{\"type\":\"marker\", \"t\":%.3f, \"k\":%d}", System.nanoTime() * 1e-9, i) )); System.out.println("Test 2"); if (i % 250 == 0) { producer.send(new ProducerRecord<String, String>( "fast-messages", String.format("{\"type\":\"marker\", \"t\":%.3f, \"k\":%d}", System.nanoTime() * 1e-9, i) )); producer.send(new ProducerRecord<String, String>( "summary-markers", String.format("{\"type\":\"other\", \"t\":%.3f, \"k\":%d}", System.nanoTime() * 1e-9, i) )); producer.flush(); System.out.println("Sent msg number " + i); } } producer.close(); } catch (Throwable throwable) { System.out.printf("%s", throwable.getStackTrace()); } finally { producer.close(); } } } Zookeeper and Kafka has already been started and created topics as indicated in the instruction: $ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic fast-messages $ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic summary-markers but when I run the compiled application $ ./kafka-example producer I get this output (on both Sandbox and Azure VM): log4j:WARN No appenders could be found for logger (org.apache.kafka.clients.producer.ProducerConfig) log4j:WARN Please initialize the log4j system properly Test 1 It gets stuck on the first producer.send line, after the Test 1 sysout log. No exception gets thrown either, it just stops at printing out that Test 1 line. When I use Eclipse to debug it using line-breaks, it stops on that line as well, apparently with nothing happening. I also tried substituting localhost, 127.0.0.1 and even the external IP address of the node for the bootstrap.server properties. The fast-messages and summary-markers directories in Kafka logs get created, but nothing written. Any idea? ... View more

I've been following the Hortonworks tutorial on all things related to Knox. tutorial-420: http://hortonworks.com/hadoop-tutorial/securing-hadoop-infrastructure-apache-knox/ tutorial-560: http://hortonworks.com/hadoop-tutorial/secure-jdbc-odbc-clients-access-hiveserver2-using-apache-knox/ Our goal is use Knox as a gateway or single point of entry for microservices that intend to connect to the cluster using REST API calls. However, my Hadoop environment is not the HDP 2.5 Sandbox but an HDP 2.5 stack built through Ambari on a single node Azure VM, so the configurations may differ and that should partially explain why the tutorials may not work. (This is a POC for a multinode cluster build) I’m testing the WebHDFS and Hive part. For Ranger, I created a temporary guest account that has global access to HDFS and Hive, as well as copying the GlobalKnoxAllow configured in the Sandbox VM (I'll take care of the more fine-grained Ranger ACL later). We didn't setup this security in LDAP mode though, just the plain Unix ACL. I also created a sample database named microservice, which I can connect to and query via beeline. From tutorial-420, Step 1: I started the Start Demo LDAP Step 2: touch /usr/hdp/current/knox-server/conf/topologies/knox_sample.xml curl -iku guest:<guestpw> -X GET 'http://<my-vm-ip>:50070/webhdfs/v1/?op=LISTSTATUS' This is what I get: HTTP/1.1 404 Not Found Cache-Control: no-cache Expires: Mon, 30 Jan 2017 07:23:17 GMT Date: Mon, 30 Jan 2017 07:23:17 GMT Pragma: no-cache Expires: Mon, 30 Jan 2017 07:23:17 GMT Date: Mon, 30 Jan 2017 07:23:17 GMT Pragma: no-cache Content-Type: application/json X-FRAME-OPTIONS: SAMEORIGIN Transfer-Encoding: chunked Server: Jetty(6.1.26.hwx) {"RemoteException":{"exception":"FileNotFoundException","javaClassName":"java.io.FileNotFoundException","message":"File /webhdfs/v1 does not exist."}} Step 3: curl -iku guest:<guestpw> -X GET 'https://<my-vm-ip>:8443/gateway/default/webhdfs/v1/?op=LISTSTATUS' This is what I get: HTTP/1.1 401 Unauthorized Date: Mon, 30 Jan 2017 07:25:51 GMT Set-Cookie: rememberMe=deleteMe; Path=/gateway/default; Max-Age=0; Expires=Sun, 29-Jan-2017 07:25:51 GMT WWW-Authenticate: BASIC realm="application" Content-Length: 0 Server: Jetty(9.2.15.v20160210) I jumped to tutorial-560: Step 1: Knox is started Step 2: In Ambari, hive.server2.transport.mode is changed to http from binary and Hive Server 2 is restarted. Step 3: SSH on my Azure VM Step 4: Connect to Hive Server 2 using beeline via Knox [root@poc2 hive]# beeline !connect jdbc:hive2:// <my-vm-ip>:8443/microservice;ssl=true;sslTrustStore=/var/lib/knox/data-2.5.0.0-1245/security/keystores/gateway.jks;trustStorePassword=knox?hive.server2.transport.mode=http;hive.server2.thrift.http.path=gateway/default/hive This is the result: Connecting to jdbc:hive2://<my-vm-ip>:8443/microservice;ssl=true;sslTrustStore=/var/lib/knox/data-2.5.0.0-1245/security/keystores/gateway.jks;trustStorePassword=knox?hive.server2.transport.mode=http;hive.server2.thrift.http.path=gateway/default/hive Enter username for jdbc:hive2://... : guest Enter password for jdbc:hive2://... : ******** 17/01/30 15:28:21 [main]: WARN jdbc.Utils: ***** JDBC param deprecation ***** 17/01/30 15:28:21 [main]: WARN jdbc.Utils: The use of hive.server2.transport.mode is deprecated. 17/01/30 15:28:21 [main]: WARN jdbc.Utils: Please use transportMode like so: jdbc:hive2://<host>:<port>/dbName;transportMode=<transport_mode_value> 17/01/30 15:28:21 [main]: WARN jdbc.Utils: ***** JDBC param deprecation ***** 17/01/30 15:28:21 [main]: WARN jdbc.Utils: The use of hive.server2.thrift.http.path is deprecated. 17/01/30 15:28:21 [main]: WARN jdbc.Utils: Please use httpPath like so: jdbc:hive2://<host>:<port>/dbName;httpPath=<http_path_value> Error: Could not create an https connection to jdbc:hive2://<my-vm-ip>:8443/microservice;ssl=true;sslTrustStore=/var/lib/knox/data-2.5.0.0-1245/security/keystores/gateway.jks;trustStorePassword=knox?hive.server2.transport.mode=http;hive.server2.thrift.http.path=gateway/default/hive. Keystore was tampered with, or password was incorrect (state=08S01,code=0) 0: jdbc:hive2://<my-vm-ip>:8443/microservi (closed)> The same thing I get even if replace hive.server2.thrift.http.path=cliserver (based from hive-site.xml). In hive-site.xml, ssl=false, so I tried substituting that on the JDBC connection URL: [root@poc2 hive]# beeline !connect jdbc:hive2:// <my-vm-ip>:8443/microservice;ssl=false;sslTrustStore=/var/lib/knox/data-2.5.0.0-1245/security/keystores/gateway.jks;trustStorePassword=knox?hive.server2.transport.mode=http;hive.server2.thrift.http.path=gateway/default/hive This is the result: Connecting to jdbc:hive2://<my-vm-ip>:8443/microservice;ssl=false;sslTrustStore=/var/lib/knox/data-2.5.0.0-1245/security/keystores/gateway.jks;trustStorePassword=knox?hive.server2.transport.mode=http;hive.server2.thrift.http.path=gateway/default/hive Enter username for jdbc:hive2://... : guest Enter password for jdbc:hive2://... : ******** 17/01/30 15:33:29 [main]: WARN jdbc.Utils: ***** JDBC param deprecation ***** 17/01/30 15:33:29 [main]: WARN jdbc.Utils: The use of hive.server2.transport.mode is deprecated. 17/01/30 15:33:29 [main]: WARN jdbc.Utils: Please use transportMode like so: jdbc:hive2://<host>:<port>/dbName;transportMode=<transport_mode_value> 17/01/30 15:33:29 [main]: WARN jdbc.Utils: ***** JDBC param deprecation ***** 17/01/30 15:33:29 [main]: WARN jdbc.Utils: The use of hive.server2.thrift.http.path is deprecated. 17/01/30 15:33:29 [main]: WARN jdbc.Utils: Please use httpPath like so: jdbc:hive2://<host>:<port>/dbName;httpPath=<http_path_value> 17/01/30 15:34:32 [main]: ERROR jdbc.HiveConnection: Error opening session org.apache.thrift.transport.TTransportException: org.apache.http.conn.ConnectTimeoutException: Connect to <my-vm-ip>:8443 [/<my-vm-ip>] failed: Connection timed out at org.apache.thrift.transport.THttpClient.flushUsingHttpClient(THttpClient.java:297) at org.apache.thrift.transport.THttpClient.flush(THttpClient.java:313) at org.apache.thrift.TServiceClient.sendBase(TServiceClient.java:73) at org.apache.thrift.TServiceClient.sendBase(TServiceClient.java:62) at org.apache.hive.service.cli.thrift.TCLIService$Client.send_OpenSession(TCLIService.java:154) at org.apache.hive.service.cli.thrift.TCLIService$Client.OpenSession(TCLIService.java:146) at org.apache.hive.jdbc.HiveConnection.openSession(HiveConnection.java:552) at org.apache.hive.jdbc.HiveConnection.<init>(HiveConnection.java:170) at org.apache.hive.jdbc.HiveDriver.connect(HiveDriver.java:105) at java.sql.DriverManager.getConnection(DriverManager.java:571) at java.sql.DriverManager.getConnection(DriverManager.java:187) at org.apache.hive.beeline.DatabaseConnection.connect(DatabaseConnection.java:146) at org.apache.hive.beeline.DatabaseConnection.getConnection(DatabaseConnection.java:211) at org.apache.hive.beeline.Commands.connect(Commands.java:1190) at org.apache.hive.beeline.Commands.connect(Commands.java:1086) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at org.apache.hive.beeline.ReflectiveCommandHandler.execute(ReflectiveCommandHandler.java:52) at org.apache.hive.beeline.BeeLine.dispatch(BeeLine.java:989) at org.apache.hive.beeline.BeeLine.execute(BeeLine.java:832) at org.apache.hive.beeline.BeeLine.begin(BeeLine.java:790) at org.apache.hive.beeline.BeeLine.mainWithInputRedirection(BeeLine.java:490) at org.apache.hive.beeline.BeeLine.main(BeeLine.java:473) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:606) at org.apache.hadoop.util.RunJar.run(RunJar.java:233) at org.apache.hadoop.util.RunJar.main(RunJar.java:148) Caused by: org.apache.http.conn.ConnectTimeoutException: Connect to <my-vm-ip>:8443 [/<my-vm-ip>] failed: Connection timed out at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:156) at org.apache.http.impl.conn.PoolingHttpClientConnectionManager.connect(PoolingHttpClientConnectionManager.java:353) at org.apache.http.impl.execchain.MainClientExec.establishRoute(MainClientExec.java:380) at org.apache.http.impl.execchain.MainClientExec.execute(MainClientExec.java:236) at org.apache.http.impl.execchain.ProtocolExec.execute(ProtocolExec.java:184) at org.apache.http.impl.execchain.RetryExec.execute(RetryExec.java:88) at org.apache.http.impl.execchain.RedirectExec.execute(RedirectExec.java:110) at org.apache.http.impl.execchain.ServiceUnavailableRetryExec.execute(ServiceUnavailableRetryExec.java:84) at org.apache.http.impl.client.InternalHttpClient.doExecute(InternalHttpClient.java:184) at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:117) at org.apache.http.impl.client.CloseableHttpClient.execute(CloseableHttpClient.java:55) at org.apache.thrift.transport.THttpClient.flushUsingHttpClient(THttpClient.java:251) ... 30 more Caused by: java.net.ConnectException: Connection timed out at java.net.PlainSocketImpl.socketConnect(Native Method) at java.net.AbstractPlainSocketImpl.doConnect(AbstractPlainSocketImpl.java:339) at java.net.AbstractPlainSocketImpl.connectToAddress(AbstractPlainSocketImpl.java:200) at java.net.AbstractPlainSocketImpl.connect(AbstractPlainSocketImpl.java:182) at java.net.SocksSocketImpl.connect(SocksSocketImpl.java:392) at java.net.Socket.connect(Socket.java:579) at org.apache.http.conn.socket.PlainConnectionSocketFactory.connectSocket(PlainConnectionSocketFactory.java:74) at org.apache.http.impl.conn.DefaultHttpClientConnectionOperator.connect(DefaultHttpClientConnectionOperator.java:141) ... 41 more Error: Could not establish connection to jdbc:hive2://<my-vm-ip>:8443/microservice;ssl=false;sslTrustStore=/var/lib/knox/data-2.5.0.0-1245/security/keystores/gateway.jks;trustStorePassword=knox?hive.server2.transport.mode=http;hive.server2.thrift.http.path=gateway/default/hive: org.apache.http.conn.ConnectTimeoutException: Connect to <my-vm-ip>:8443 [/<my-vm-ip>] failed: Connection timed out (state=08S01,code=0) 0: jdbc:hive2://<my-vm-ip>:8443/microservi (closed)> Again I tried replacing hive.server2.thrift.http.path=cliserver and I get the same result. Does anyone here able to configure to Knox correctly and working? ... View more

I've noticed that this version of HDP 2.5 Sandbox uses Docker container. When the VM launches I can login as root (pw: hadoop) on the default console. However, when I use ssh on other terminal, like Git Bash and Putty (which I am more comfortable using), I constantly get the message 'Permission Denied' even though I am keying-in same password. ssh root@<my-ip> -p 2222 Another thing I noticed is the hadoop commands (hive, pig, oozie, etc.) are missing and the /usr/hdp path are also not in the Linux path. I really need to get into the terminal. ... View more