I'm working with Spark Structured Streaming. I want to pass config file to my spark application. This configuration file hosted in HDFS . For example; spark_job.conf (HOCON) spark {   appName: "",   master: "",   shuffle.size: 4   etc..  } kafkaSource {   servers: "",   topic: "",   etc.. } redisSink {   host: "",   port: 999,   timeout: 2000,   checkpointLocation: "hdfs location",   etc.. } How can I pass it to Spark Application? How can I read this file( hosted HDFS ) in Spark? ... View more

I've seen mapgroupswithstate example in Databricks blog . (https://databricks.com/blog/2017/10/17/arbitrary-stateful-processing-in-apache-sparks-structured-streaming.html). But I wonder what happens if I use "var" instead of "val" in mapgroupswithstate function. For example, databricks shared this example above link; case class UserState(user:String, var activity:String, var start:java.sql.Timestamp, var end:java.sql.Timestamp) They defined State fields with "var". And mapgroupswithstate function like this; def updateAcrossEvents(user:String,     inputs: Iterator[InputRow],     oldState: GroupState[UserState]):UserState = {       var state:UserState = if (oldState.exists) oldState.get else UserState(user,         "",                                                                                  new java.sql.Timestamp(6284160000000L),                                                                                     new java.sql.Timestamp(6284160L)    )   // we simply specify an old date that we can compare against and             // immediately update based on the values in our data   for (input <- inputs) {     state = updateUserStateWithEvent(state, input)     oldState.update(state)   }   state } Also, in this example, state was defined as "var" and updated with each event. I know "var" vs "val" in Scala. Also, "val" is important for functional programming, and it safe. But I don't know what happens if I use "var" like above example? For above example, does using "var" create a problem? ... View more

I'm trying to convert json to json using JoltTransformRecord. I'm getting Json data from ForkRecord processor. When I try split "received json" using SplitRecord with fork output schema(I'm using fork output schema for JsonTreeReader of SplitRecord), data successfully splitted. But, when I try to convert Json using JoltTransformRecord, I get error. My output schema of JsonRecordSetWriter of JoltTransformRecord is correct for in my opinion. But this input schema of JoltTransformRecord is the same the output schema of the ForkRecord processor. Also, I have tried Jolt specification in Jolt transform demo, and result is correct. Then I have created ouput schema of JoltTransformRecord by this result. I don't understand why I'm gettin this error. I have tried a lot of way, schema etc.. But I didn't anything. XML file defined below, contains my flow with schemas. test_transform.xml Details logs in my nifi-app.log; org.apache.nifi.serialization.record.util.IllegalTypeConversionException: Cannot convert value of class [Ljava.lang.Object; because the type is not supported     at org.apache.nifi.serialization.record.util.DataTypeUtils.convertRecordFieldtoObject(DataTypeUtils.java:460)     at org.apache.nifi.serialization.record.util.DataTypeUtils.convertRecordMapToJavaMap(DataTypeUtils.java:472)     at org.apache.nifi.serialization.record.util.DataTypeUtils.convertRecordFieldtoObject(DataTypeUtils.java:444)     at org.apache.nifi.processors.jolt.record.JoltTransformRecord.onTrigger(JoltTransformRecord.java:318) Please, help. This is so imporant for me. @Bryan Bende @Matt Burgess @Shu ... View more

I receive JSON data from kafka with from_json() method. It expects schema from me. My JSON structure like this;   {     "Items": {         "key1": [             {                 "id": "",                 "name": "",                 "val": ""             }         ],         "key2": [             {                 "id": "",                "name": "",                 "val": ""             }         ],         "key3": [             {                 "id": "",                 "name": "",                 "val": ""             }         ]     } }   Key1, Key2, Key3 are dynamic. So, they may be changed. For example, another json is; {     "Items": {         "hortoworks": [             {                 "id": "",                 "name": "",                 "val": ""             }         ],         "community": [             {                 "id": "",                "name": "",                 "val": ""             }         ],         "question": [             {                 "id": "",                 "name": "",                 "val": ""             }         ]     } } These key names are unknown. But "id-name-val" fields inside these keys are the same. I must define a json schema for read data from Kafka in Spark Structured Streaming. How can I do this? ... View more

I get different XMLs from web services. I want to convert this XML to Json, but structure must be changed. For example, I have XML structure like this; <root>     <A attr="attr1">VAL</A>     <B attr="attr2">VAL</B> </root> And result of JSON that I want;; "root": {     "Items": [         {             "tag_name": "A",             "attr": "attr1",             "value": "VAL"         },         {             "tag_name": "B",             "attr": "attr2",             "value": "VAL"         }     ]  } How can I do this in Nifi? ConvertRecord etc.. Also, how should read and write schema be for this if record based processors may be used? ... View more

I am getting about 1000 xml per second. Each XML about 50-60 KB. I think of using SplitXML and EvaluateXPath for XML parsing. But I don't know that how memory usage and performance will be. Does these processors load entire xml to memory? ... View more

I have XML data like this; <All> <Header creationTime="2019-02-04T17:10:53Z"/> <Specific> <New> ..... </New> </Specific> </All> But sometimes, XML is coming like this; <All> <Header creationTime="2019-02-04T17:10:53Z"/> <Specific/> </All> I want to check that if received XML contains "<Specific/>" . How can I do that? ... View more

I will send request to ListenHttp(running on each node) using PostHTTP(running on only primary node). For this, i am getting nodes address from nifi rest-api(/controller/cluster). And i prepare URL like this; http://nodeAddress:port/contentListener Can I successfully send request to ListenHttp processors in each node with PostHttp using this URL? ... View more

I want to get node details from nifi rest-api in primary node. For this, I can use "192.xxx.xx.xx:8080" address. But, what happens if primary node is down? I can not access the rest api using above address, after new primary node will be elected. Can I use "localhost:8080" address for this purpose for all nifi nodes? Does this work? ... View more