@hem lohani Create column using withColumn function with literal value as 12. Use month column as partitionby column and use insertInto table. df.withColumn("month",lit(12)).write.mode("<append or overwrite>").partitionBy("month").insertInto("<hive_table_name>") (or) Using SQL query df.createOrReplaceTempView("temp_table") spark.sql("insert into <partition_table> partition(`month`=12) select * from <temp_table>") - If the answer is helpful to resolve the issue, Login and Click on Accept button below to close this thread.This will help other community users to find answers quickly 🙂 ... View more

@hem lohani Try with below syntax df.write.mode("<append or overwrite>").partitionBy("<partition_cols>").insertInto("<hive_table_name>") ... View more

@Sidhartha Bobby To run exchange partition successfully, Your destination table base.customers cannot contain the partition that you are exchanging. But base.customers table is already have the partition(source_name=ORACLE) that you are exchanging. Resolution: Delete the existing partition in destination table and run the exchange partition command again. (or) Exchange the partition that is already not exists in destination table. If you want to just append the data to Destination table then run insert into base.customers table by selecting from stg.customers_testcontrol_staging table. From HiveDocs: Constraints for Hive Exchange partitions: The destination table cannot contain the partition to be exchanged. The operation fails in the presence of an index. Exchange partition is not allowed with transactional tables either as source or destination. Alternatively, use LOAD DATA or INSERT OVERWRITE commands to move partitions across transactional tables. This command requires both the source and destination table names to have the same table schema. If the schemas are different, the following exception is thrown: The tables have different schemas. Their partitions cannot be exchanged ... View more

@Khaja Mohd Use GetFile processor as this processor allows to keep specific patterns. Refer this link for configuring GetFile processor to use with File Filter patterns. ... View more

@manohar ghanta Option-1: You can do .coalesce(n) (no shuffle will happen) on your dataframe and then use .option("maxRecordsPerFile",n) to control the number of records written in each file. Option-2: Using spark.sql.shuffle.partitions=n this option is used to control the number of shuffles happens. Then use df.sort("<col_name>").write.etc will create exactly the number of files that we mentioned for shuffle.partitions . Option-3: Hive: Once the spark job is done then trigger hive job insert overwrite by selecting the same table and use sortby,distributedby,clusteredby and set the all hive configurations that you have mentioned in the question. Insert overwrite table select * from table sort by <col1> distributed by <col2> Option-4: Hive: If you have ORC table then schedule concatenate job to run periodically alter table <table_name> concatenate; If none of the methods seems to be feasible solutions then .repartition(n) will be the way to go as this will take extra overhead but we are going to end up ~evenly sized filesin HDFS and boost up the performance while reading these files from hive/spark. - If the answer is helpful to resolve the issue, Login and Click on Accept button below to close this thread.This will help other community users to find answers quickly 🙂 ... View more

@Adarsh R You can try using InferAvroSchema processor with SchemaOutputDestination as flowfile-attribute. Then processor adds "inferred.avro.schema" attribute to the flowfile use this attribute in your Record oriented processors. Refer this and this links for more details regards to usage and challenges with the inferavroschema processor. ... View more

@Piotr Grzegorski Try using ListHDFS + FetchHDFS processors. You can simulate MoveHDFS processor with the below Flow: ListHDFS //list all the files in HDFS directory RouteOnAttribute //Use nifi expression language to filter out the required files FetchHDFS //fetch the files from HDFS PutHDFS //put the files into HDFS directory. DeleteHDFS //delete the file from HDFS directory that are pulled from FetchHDFS - If the answer is helpful to resolve the issue, Login and Click on Accept button below to close this thread.This will help other community users to find answers quickly 🙂 ... View more

@Prathamesh H It seems AccessControlException on /ranger/audit directory as directory have hdfs user access but hive user is trying to write to .log in the direcrtory. Could you change permissions on this directory as per this link and try to insert data into Hive table? ... View more

@Krishna Murthy In convertRecord processor define xml reader and csv writer controller services. In CSV writer controller service define avro schema only the required fields you need to be in csv file. ... View more

@srini This is Hive bug reported here HIVE-2927. Possible Workarounds would be: Hive >=1.3.+ Then use replace function in hive with get_json_object Example: hive> select get_json_object(replace(jsn,'@',''),"$.date") from (select string('{"name":"jai","@date":"2015-06-15"}')jsn)t; 2015-06-15 (or) hive> select get_json_object(replace(jsn,'@date','date'),"$.date") from (select string('{"name":"jai","@date":"2015-06-15"}')jsn)t; 2015-06-15 Hive <1.3: Use regexp_replace function: hive> select get_json_object(regexp_replace(jsn,'@',''),"$.date") from (select string('{"name":"jai","@date":"2015-06-15"}')jsn)t; 2015-06-15 (or) hive> select get_json_object(regexp_replace(jsn,'@date','date'),"$.date") from (select string('{"name":"jai","@date":"2015-06-15"}')jsn)t; 2015-06-15 -->Final query would be: hive> select get_json_object(jsn,'$.name'),get_json_object(regexp_replace(jsn,'@',''),"$.date") from (select string('{"name":"jai","@date":"2015-06-15"}')jsn)t; jai     2015-06-15 - If the answer is helpful to resolve the issue, Login and Click on Accept button below to close this thread.This will help other community users to find answers quickly 🙂 ... View more

@Jim Barnett Not able to recreate the same scenario on my end(using hive 1.2.1). Wondering how count(distinct key) is generated just one row of count as result ,but in your case query is giving results as select key,count(*) from default.dummy_data group by key; ----- Check is there any data that already exists in hdfs directory that table is pointing to, If yes clear off the data from the hdfs directory and recreate the table again. Example: hive> CREATE TABLE default.dummy_data     > AS     > SELECT row_nbr as key     > FROM (     > SELECT row_number() OVER (partition by '1') as row_nbr     > FROM (     > select explode(split(repeat("x,", 1000000-1), ",")) -- 1,024 distinct     > ) AS x     > ) AS y; hive> select count(distinct key) c from default.dummy_data; 1000000 hive> select key,count(*)cnt from default.dummy_data group by key order by cnt desc limit 10; --ordering desc and limiting 10 Result: key     cnt 10      1 9       1 1000000 1 7       1 6       1 5       1 4       1 3       1 999999  1 1       1 ... View more

@Juhyeon Yun Configure directory as /user/paxata/job and keep recursive sub directories to true will list all the files in the sub directories. Similar question reported in this HCC thread. ... View more

@Bora Özkan One thing i observed in the article is author created hive external table but PutHiveStreaming expects internal table. Create hive internal table CREATE TABLE OLYMPICS(CITY STRING,EDITION INT,SPORT STRING,SUB_SPORT STRING,ATHLETE STRING,COUNTRY STRING,GENDER STRING,EVENT STRING,EVENT_GENDER STRING,MEDAL STRING) CLUSTERED BY (EDITION)INTO 3 BUCKETS ROW FORMAT DELIMITED STORED AS ORC  LOCATION  '/user/hive/olympics' TBLPROPERTIES('transactional'='true'); -> PutHiveStreaming processor expects input data in Avro format. Use ConvertRecord processor to convert data from other formats(csv..etc) to AVRO format. Then feed the avro format data to PutHiveStreaming processor. ... View more

@Karthikeyyan Subbiah I think your attribute is not coming as attribute of the flowfile, if your attribute is in content of flowfile then UpdateAttribute processor won't work. We need to use UpdateRecord processor for that and give the path as /<attribute_name> and add your logic to get the desired format of timestamp. ... View more

@Veera Pavan This job will work fine in Hive but in Spark follow these steps: write the data to temporary table first. then select from temporary table insert overwrite the final table. Check this similar thread regards to similar case. - If the answer is helpful to resolve the issue, Login and Click on Accept button below to close this thread.This will help other community users to find answers quickly 🙂 ... View more

@Carlo Bustos Could you confirm that DistributedMapCacheClientService is defined and enabled ? Please refer this and this links for defining distributed map cache service. ... View more

@Veerendra Nath Jasthi In UpdateAttribute add new attribute as ts value as ${now():format("yyyy_MM_dd_HH_mm_ss_SSS")} Example: ts attribute will have value as 2019_06_12_20_42_26_762 Then in PutHDFS processor configure directory as /<path>/${ts} (or) You can skip UpdateAttribute processor and directly use directory name as /<path>/${now():format("yyyy_MM_dd_HH_mm_ss_SSS")} In PutHDFS processor. This will create a directory in HDFS with current timestamp value. You can change the format of the timestamp using NiFi expression language. ... View more

@James Willson Try with below spec: [{     "operation": "shift",     "spec": {       "rows": {         "*": {           "value": "[&1].date",           "data": {             "*": "[#3].data"           }         }       }     } } ] Gives output that you are looking for: [ {   "date" : "00:00 2019-06-03",   "data" : 120 }, {   "date" : "05:00 2019-06-08",   "data" : 98 }, {   "date" : "23:00 2019-06-09",   "data" : 172 } ] - If the answer is helpful to resolve the issue, Login and Click on Accept button below to close this thread.This will help other community users to find answers quickly 🙂 ... View more

@jingyong zou The issue is with the flowfile format that is passed through the processor. As ConvertAvroToJson processor accepts only Avro format but i think you are passing Json format to the processor which is causing java.io.IOException: Not a data file error. ... View more