Q&A

has anyone developed data flows that ingest document(csv+blobs) into ECM system like Nuxeo

We are using ListSFTP and FetchSFTP to get the files from a vendor SFTP location. There are no explicit delete commands from our end to delete the file . However the file gets deleted and the vendor is pushing it on us saying we are having some process to delete the files . Will FetchSFTP delete the files after its downloaded from the Vendor SFTP server ?

For the below given input, how do i move the name value pairs(installDate and status) to the upper level{"header": {"version": "1.0","timeStamp": "1670484663189","activityId": "256388257993155783","action": "create","refreshId": "null"},"resource": {"drniId": "256383859946641699","@type": "LogicalCrossConnection","subtype": "Backplane Connection","name": "Backplane Connection","resourceCharacteristic": [{"name": "tmf639_dataSource_ingestion_uuid","value": "256383859946641699","@type": "string"},{"name": "p16Latest","value": true,"@type": "boolean"},{"name": "installDate","value": "202303323","@type": "string"},{"name": "status","value": "In Service","@type": "string"}],"resourceRelationship": [{"drniId": "256387639517862117","@type": "LogicalInterface:HAS_CONNECTION_COMPONENT:OUT","subtype": "ODU2e","name": "ODU2e.1-8","relationDrniId": "256383859946641700","resourceRelationshipCharacteristic": [{"name": "p16Latest","value": true,"@type": "boolean"},{"name": "sectionNumber","value": 0,

Hi Jolt Experts,Iam looking to use the Jolt transformation in NIFI and looking for a help.{"header": {"timeStamp": "1681441547144","activityId": "5142917973836294867"},"resource": {"drniId": "261968731599587823","@type": "Order","subtype": "Retail","name": "CCCCC","resourceCharacteristic": [{"propname": "aaa","value": "P_1655442"},{"name": "latest","value": [11,200,300,16,2,3]}]}} To have an output like{"records" : [ {"timeStamp" : "1681441547144","activityId" : "5142917973836294867","drniId" : "261968731599587823","type" : "Order","subtype" : "Retail","name" : "CCCCC","propname" : "aaa","value" : "P_1655442"}, {"timeStamp" : "1681441547144","activityId" : "5142917973836294867","drniId" : "261968731599587823","type" : "Order","subtype" : "Retail","name" : "CCCCC","propname" : "latest","value" : "11,200,300,16,2,3"} ]} Basically copying all the top header and resource into to new by flattening it. Any help appreciated.  

I am running a cluster with tlswhat commands I used to create tls, perhaps something was not indicated in them:.I have crt and key my companyopenssl pkcs12 -export -in /etc/nifi/certs/wibe.t.crt -inkey /etc/nifi/certs/wibe.t.key -out /etc/nifi/certs/pkcs12_file.p12 -name nifi_alias -CAfile /etc/pki/CA.pem -caname root -password pass:qwerty231keytool -importkeystore -deststorepass "keystore_password" -destkeypass "keystore_password" -destkeystore /etc/nifi/certs/nifi_keystore.jks -srckeystore pkcs12_file.p12 -srcstoretype PKCS12 -srcstorepass "pkcs12_password" -alias nifi_alias keytool -import -trustcacerts -alias root -file /etc/pki/CA.pem -noprompt -keystore /etc/nifi/certs/nifi_truststore.jks -storepass "truststore_password"at first I could not collect a quorum, then this problem went away and the cluster chose a leader, work is done with external zookeepers on three serversno errors, only warning2023-05-04 16:06:34,116 WARN [main] o.a.nifi.controller.StandardFlowService Failed

I have an exception thrown when trying to write a data frame to kudu of size 524GB.After calling this writing part:  df.write.format("org.apache.kudu.spark.kudu").option( "kudu.master", "master1:port,master2:port,master3:port",).option("kudu.table", f"impala::{schema}.{table}").mode("append").save() this exception is thrown: java.lang.RuntimeException: PendingErrors overflowed. Failed to write at least 1000 rows to Kudu; Sample errors: Timed out: cannot complete before timeout: Batch{operations=1000, tablet="0bc1e2a497ab4306b6861f81dc678d9f" [0x00000002, 0x00000003), ignoredErrors=[], rpc=KuduRpc(method=Write, tablet=0bc1e2a497ab4306b6861f81dc678d9f, attempt=26, TimeoutTracker(timeout=30000, elapsed=29585), Trace Summary(29585 ms): Sent(26), Received(26), Delayed(26), MasterRefresh(0), AuthRefresh(0), Truncated: false here is the yarn.log for this spark job that throws the errorYarn log for a spark job Here is the er

I have read the descriptions of both these two Nifi processors and I am unable to understand the difference. The description seems to be almost same for both. It will be nice if Nifi docs description provided some note on the difference.  Main difference seems to be:QueryDatabaseTable: Query result will be converted to Avro format.QueryDatabaseTableRecord: Query result will be converted to the format specified by the record writer. So there is a separate Record Writer field in the properties. 

Hello. Well, I have a nifi flow that is triggered by a ListFile processor. ListFile creates an empty content flowfile and sends it to success queue.I want to write some fixed content ( say "id,description,date" ) on this empty flowfile. The goal is to write an empty csv with that header in HDFS  as first action of the flow.So I was trying to use ReplaceText for writing the content to the empty flowfile.The flow looks like this    ListFile ---> ReplaceText (see image below)ReplaceText has the following settings: Replacement Strategy = Always Replace, Replacement Value = id,description,dateAt first I thougt that ReplaceText could fail cause incoming flowfile had no content.But ReplaceText sent the flowfile to success BUT NO CONTENT WAS WRITTEN (It is still empty)In the image below you can see that Queued Bytes are 0  Setting Bulletin Level to "DEBUG" just throws the following message : "ReplaceText[id=501319d3-354d-1482-6c84-d176639be85f] Transferred FlowFile[

 Dears,    We are working on nifi flows. Creating Processor groups in Normal web version of NIFI.  But as per development , We need to move these developed processor groups to Licensed              version of NIFI for real Development.  But we do not have proper idea that how we get licensed version of nifi and how to export to all  development to that version? Can you kindly suggest me an approach to achieve this Thank you             

Hi all,We are trying to configure a flow in Nifi where failed sql queries are saved after some retries to local file for further analysis.I already see the following links related to error handling in nifihttps://youtu.be/nrxv65J4Ctg by Steven Koonhttps://insights.1904labs.com/blog/2020-11-12-creating-an-error-retry-framework-in-nifi-part-2But those articles do not refer to PutSql processor, I'm not sure if the case apply. The failed query in our scenario using PutSQL processor is related to wrong registry information on origin database, giving error inserting a text value on a type bigint column on destination sql server database.This is an example sql query used for inserts on destination database:INSERT INTO destination (id, name, description, group_id) VALUES (100, 'Bob', 'Example query', '2151')We have a wrong registry on origin database in where the group_id field have value 'null' getting the following exception on nifi logs:org.apache.nifi.processor.exception.ProcessE

HI all, I'm getting the following error when trying to launch pyspark. Please help !! Thanks in advance.C:\Users\abhay>pysparkPython 3.10.3 (tags/v3.10.3:a342a49, Mar 16 2022, 13:07:40) [MSC v.1929 64 bit (AMD64)] on win32Type "help", "copyright", "credits" or "license" for more information.WARNING: An illegal reflective access operation has occurredWARNING: Illegal reflective access by org.apache.spark.unsafe.Platform (file:/D:/demo/spark-3.2.1/jars/spark-unsafe_2.12-3.2.1.jar) to constructor java.nio.DirectByteBuffer(long,int)WARNING: Please consider reporting this to the maintainers of org.apache.spark.unsafe.PlatformWARNING: Use --illegal-access=warn to enable warnings of further illegal reflective access operationsWARNING: All illegal access operations will be denied in a future releaseUsing Spark's default log4j profile: org/apache/spark/log4j-defaults.propertiesSetting default log level to "WARN".To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLo

When I query impala like "select count(1) from t", the impala daemon is correct。but when I change query  "select count(distinct(xxx)) from t", the impala daemon unexpected exit.  Error Log:llvm-codegen.cc:137] dc448ef835296647:2da12cb500000001] LLVM hit fatal error: Cannot select: intrinsic %llvm.x86.sse42.crc32.32.32

Hello,I would like to connect my Airflow application with my Apache Nifi application though the Apache Nifi API. The aim is to let the Airflow orchestrate my entire flow.My Apache Nifi authentication is made via the OpenId protocol on a Keycloak application so I first get a token via a call to the auth/realms/myrealms/protocol/openid-connect/token and then with the generated JWT, I call the nifi-api to catch some processors, passing the JWT as a bearer token, I have got the following error : Bearer error="invalid_token", error_description="An error occurred while attempting to decode the Jwt: Signed JWT rejected: Another algorithm expected, or no matching key(s) found", error_uri="https://tools.ietf.org/html/rfc6750#section-3.1" I tried the same curl call to the processor, using the token I get when I connect with the UserInterface, and I get a success. I supposed it may be due to the internal user policy of Nifi who rejecte me so I added username and

I need to List and Fetch specific files on S3 from April 2022 to December 2022. I was hoping to narrow the listing using the Minimum and Maximum Object Age properties as specificed in the processor documentation.However, I can;t find the Maximum Object Age property?Anybody knows how to use it?Any other strategy that might be useful? Listing all the files in the prefix takes forever as it contains a huge amount of data. Thanks.

We have done the NiFi upgradation from 1.18.0 to 1.19.1 and we also upgrade the java version from 8 to 11 and after upgradation we observe nifi using more memory than allocated and caching build in the nifi. we allocate memory to nifi as java.arg.2=-Xms1g and java.arg.3=-Xmx50g. we are using RAM-64 GB and Core-8 and the solid-state high-speed hard disk is 2T. OS is CentOS7.9, ARM64 hardware architecture. we uncomment the Garbage collector from bootstarp.conf but this is not useful for us. we keep the databasr repository, content repository, provenance repository, and flowfile repository in different disks.I need a help to resolve the issue nifi-app.log for reference id=1682607799237-15502, container=contS1R1, section=142], offset=605303, length=7664],offset=0,name=47ffd206-8189-4fbb-9d19-2b82113da956,size=7664], StandardFlowFileRecord[uuid=a06c6ecd-167f-41d6-9ec5-1bcadc37fcf6,claim=StandardContentClaim [resourceClaim=StandardResourceClaim[id=1682607797216-15364, container=con

I am trying to load a simple file to HDP Hadoop cluster using HDFS client and I got the following exception. Exception in thread "main" org.apache.hadoop.ipc.RemoteException(java.io.IOException): File /home/maria_dev/read_write_hdfs_example.txt could only be replicated to 0 nodes instead of minReplication (=1). There are 1 datanode(s) running and 1 node(s) are excluded in this operation.I looked into the namenode logs and enabled debug log level for NetworkTopology and BlockPlacementPolicy components. After enabling logs, I found that the data node 172.18.0.2:50010 is  being excluded and since I am running only one datanode, it is unable to find second replica. 2023-04-28 06:00:22,188 DEBUG net.NetworkTopology (NetworkTopology.java:chooseRandom(780)) - Choosing random from 1 available nodes on node /default-rack, scope=/default-rack, excludedScope=null, excludeNodes=[] 2023-04-28 06:00:22,188 DEBUG net.NetworkTopology (NetworkTopology.java:chooseRandom(796)) - choos

I have edited the install.properties file for the trino plugin and am using this repository https://github.com/aakashnand/trino-ranger-demo.git  I edited the  XAAUDIT.ELASTICSEARCH.URL to my IP and XAAUDIT.ELASTICSEARCH.PORT to my port, and I am using similar configuration for apache hive plugin too and its working there but for trino I get this error when I perform a task and it tried to send audit message.     2022-11-29T09:48:32.780Z INFO org.apache.ranger.audit.queue.AuditBatchQueue1 stdout INFO - {"repoType":8,"repo":"apache_trino_onprem_8","reqUser":"testuser","evtTime":"2022-11-29 09:48:32.543","resource":"tpch/sf1/nation/comment","resType":"column","action":"select","result":1,"agent":"trino","policy":141,"enforcer":"ranger-acl","agentHost":"my-localhost-trino","logType":"RangerAudit","id":"718df890-2d9c-4641-96fe-c38fd58f8a20-37","seq_num":75,"event_count":1,"event_dur_ms":0,"tags":[],"cluster_name":"","policy_versio

My process looks like thisConsumerKafka 2.6 --- JoltTransformJSON---- ConvertRecord ----MergeRecord---PutHDFSmy task is i get data from kafka in jsion format i need to write large files to ndfs in parquet format, but i get small files of 10kb each i would like to receive files of at least 100mb (changed settings in merge always sends small files) and i get flow files, but you need parquet what's wrong.1- The file must be large2- file must be parquet

I am trying to add a node in my cluster. However, after trying to register the new node to the cluster. I am getting the error below while registering it.Ambari version. 2.3.4.7OS : Ubuntu 14.04ERROR 2020-04-15 14:10:08,335 NetUtil.py:84 - [Errno 8] _ssl.c:510: EOF occurred in violation of protocolERROR 2020-04-15 14:10:08,336 NetUtil.py:85 - SSLError: Failed to connect. Please check openssl library versions.Refer to: https://bugzilla.redhat.com/show_bug.cgi?id=1022468 for more details.WARNING 2020-04-15 14:10:08,336 NetUtil.py:112 - Server at https://xxxxxxx:8440 is not reachable, sleeping for 10 seconds...', None)Troubleshooting:- Python version is the same new node and the cluster. I am using python 2.6.- Added also in ambari-client.ini the security protocol TLSv1.- Java version is also the same on all node like the new node. I am using java version "1.8.0_92".- DNS is also properly set. Got positive result when digging fqdn is resolving in the DNS server.- Host file is also pr

Hi. I am trying to write data to hadoop cluster running in HDP 2.6.5 from a local nifi instance. But I receive the following error on writeorg.apache.hadoop.ipc.RemoteException(java.io.IOException): File /home/maria_dev/.2d0282a9-8c6a-4bde-beb1-82eb1b061941 could only be replicated to 0 nodes instead of minReplication (=1).  There are 1 datanode(s) running and 1 node(s) are excluded in this operation.In the HDFS data node logs, I see the below error logs2023-04-18 12:05:13,386 WARN blockmanagement.BlockPlacementPolicy (BlockPlacementPolicyDefault.java:chooseTarget(385)) - Failed to place enough replicas, still in need of 1 to reach 1 (unavailableStorages=[], storagePolicy=BlockStoragePolicy{HOT:7, storageTypes=[DISK], creationFallbacks=[], replicationFallbacks=[ARCHIVE]}, newBlock=true) For more information, please enable DEBUG log level on org.apache.hadoop.hdfs.server.blockmanagement.BlockPlacementPolicy and org.apache.hadoop.net.NetworkTopology Have validated follow things

Hi, I have flow file with an Avro record. Is there any chance to extract one field value from this record to attribute without converting record to Json format and without using the groovy script. I need to extract field value from content to attribute for using it as a part of cache identifier in the further processor.  Thank you.

Hello all,I'm setting up a process in NIFI to download some files from S3 bucket. The process contain a ListS3, UpdateAttribute, FetchS3Object and PutHDFS on a 3 nodes cluster environment. The ListS3 retrieve the files but the files get stuck in the queue before the FetchS3Object and will not move past the FetchS3Object.  I don't received any error message and the files would just sit there. Any reason why the files are not moving to the FetchS3Object processor?  I have all default settings set on both ListS3 and FetchS3Object.Thanks!

I am trying to insert a flow file into MongoDb which has a createdAt date key record as an attribute.When I inserted that flow file into MongoDb using PutMongo processor, it saves the "createdAt" attribute as a string. I want this to be saved as an ISO date object in mongoDB.When I inserted flow file by sending "2020-08-26T04:00:00.000Z" PutMongo Processor inserts as a string.When I sent like this “ISODate("2020-08-26T04:00:00.000Z”)” it also inserts same string as it is into mongo.When I tried like this ISODate("2020-08-26T04:00:00.000Z”) without double quotation it throws error invalid object. I need output as : {“createdAt”: ISODate("2020-08-26T04:00:00.000Z”)} Kindly help if there is any way to do so.

Hi All,I am getting error when I restart the NiFi server and run the CaptureChangeMySql processor or re-configure any processor and start the processor. It works fine when I creates  CaptureChangeMySql processor and run it. When I restart the server this I start getting below error. It works fine again when processor is recreated freshly. Any Idea how I can fix this issue. 2022-07-07 15:10:58,307 ERROR [Timer-Driven Process Thread-8] o.a.n.c.m.processors.CaptureChangeMySQL CaptureChangeMySQL[id=d7f0536f-0181-1000-db94-5950dddc0439] Processing failedorg.apache.nifi.processor.exception.ProcessException: java.io.IOException: COMMIT event received while not processing a transaction (i.e. no corresponding BEGIN event). This could indicate that your binlog position is invalid.at 

How to know when NiFi has done processing with all the flowfiles in the processor group? Also is there any way to notify external API?