@Alvin Jin To answer your question about which processors to use: it depends on what you want to do with the whole CSV file. Your question only mentions splitting and ignoring the header, the CSVReader takes care of that. The record-aware processors in NiFi 1.3.0 include: ConsumeKafkaRecord_0_10: Gets messages from a Kafka topic, bundles into a single flow file instead of one per message ConvertRecord: Converts records from one data format to another (Avro to JSON, e.g.) LookupRecord: Uses fields from a record to lookup a value, which can be added back to the record PartitionRecord: Groups "like" records (based on user-provided criteria) into individual flow files PublishKafkaRecord_0_10: Posts messages to a Kafka topic PutDatabaseRecord: Executes a specified operation (INSERT, UPDATE, DELETE, e.g.) on a database for each record in a flow file PutElasticsearchHttpRecord: Executes a specified operation ("index", e.g.) on an Elasticsearch cluster for each record in a flow file QueryRecord: execute SQL queries on fields from the records. This can be used to filter, aggregate, etc. SplitRecord: Splits records into smaller flow files. Usually only used when downstream processors are not record-aware UpdateRecord: Updates field(s) in each record of a flow file Also I wanted to mention, if for some reason all your CSV columns are strings, you can set "Schema Access Strategy to "Use String Fields From Header", and then you don't need a schema or schema registry. Otherwise if you want to provide a schema, you're not required to use a schema registry, you can just paste your schema into the Schema Text property. and set "Schema Access Strategy" to "Use Schema Text Property". ... View more

As of NiFi 1.3.0, you can use UpdateRecord for this. If your incoming field name is "createdOn", you can add a user-defined property named "/createdOn" whose value is the following: ${field.value:toDate('yyyy-mm-dd HH:mm:ss.SSS'):toNumber()} Note that you may need to change the type of createdOn from String (in the Reader's schema) to Long (in the Writer's schema). ... View more

You can try a thread dump (with jstack or nifi.sh dump) while it is waiting to shut down, you may be able to spot the culprit in the output. ... View more

I tested this with Arabic characters in my text field, and it worked fine. You're saying you still get the error when using my suggested lines? ... View more

The documentation says "The Expression Language allows single quotes and double quotes to be used interchangeably". Try double-quotes in your EL expression. ... View more

You can accomplish this with ExecuteScript, the following example uses Groovy as the language and Jayway's JsonPath as the library for JSONPath parsing. First I had to download 3 JARs (json-path and its required transitive dependencies) into a directory: Then I set the Module Directory property to the path of this directory: Note that I have also added a dynamic property, whose name will become an attribute and whose value supports Expression Language and (after evaluation) should contain a JSONPath expression used to retrieve the value from the content of the JSON flow file. The Script Body is the following Groovy script: import com.jayway.jsonpath.* def flowFile = session.get() if(!flowFile) return def inputStream = session.read(flowFile) def json = JsonPath.parse(inputStream) inputStream.close() context.properties.findAll {p,s -> p.dynamic}.each {pd, name -> def prop = context.getProperty(pd) try { flowFile = session.putAttribute(flowFile, pd.name, json.read(prop.evaluateAttributeExpressions(flowFile).value)) } catch (e) { log.error("Error evaluating JSONPath expression in property $name: ${prop?.value} , ignoring...", e) } } session.transfer(flowFile, REL_SUCCESS) I tested this with a GenerateFlowFile: After the ExecuteScript transfers the flow file, it has the desired attribute name/value: This should work with any number of attributes/JSONPaths per flow file. In addition, I have written NIFI-4100 to cover the improvement to the EvaluateJsonPath processor to support Expression Language. ... View more

Also, as of NiFi 1.3.0 / HDF 3.0.0, GenerateTableFetch accepts incoming connections/flow files, so you can use ListDatabaseTables -> GenerateTableFetch -> RPG -> Input Port -> ExecuteSQL to fully distribute the fetching of batches of rows across your NiFi cluster. The RPG -> Input Port part is optional and only used on a cluster if you want to fetch rows in parallel. ... View more