Hi @Raj Zalavadia As described in SPARK-16996 and SPARK-15348, Spark currently doesn't support Hive ACID ( v1 (Hive 1.XX) or v2 (3.XX) ) To circumvent that you can use the Hive Warewhouse connector. It will create the necessary link between the 2 components, by getting Spark to connect via Hive Server2. I'm not sure if it's directly bundled into HDI (should be). In any case, it's available publicly at : https://github.com/hortonworks/hive-warehouse-connector-release/tree/HDP-3.0.1.10-7-tag You'll find the documentation here : https://docs.hortonworks.com/HDPDocuments/HDP3/HDP-3.1.0/integrating-hive/content/hive_hivewarehouseconnector_for_handling_apache_spark_data.html Here's another HCC article that gives you a concrete example on how to use it. : https://community.hortonworks.com/articles/223626/integrating-apache-hive-with-apache-spark-hive-war.html To get you started, here's a quick example of how to use it : 1. The Hive Warehouse connector must be given as a dependency to spark spark.jars=[path to the Hive warehouse connector] usually : /usr/hdp/current/hive_warehouse_connector/hive-warehouse-connector-assembly-[Build version].jar 2. It also requiers a few more configuration - basically describing where the Hive meta store and Hive Server2 instances reside spark.datasource.hive.warehouse.metastoreUri=thrift://[YOUR METASTORE URI]:9083 spark.hadoop.hive.llap.daemon.service.hosts=@llap0 spark.hadoop.hive.zookeeper.quorum=[YOUR HIVE ZOOKEEPER QUORUM]:2181 spark.sql.hive.hiveserver2t.jdbc.url=[YOUR HIVE LLAP JDBC URL] These can be passed as spark conf items --conf spark.hadoop.hive.llap.daemon.service.hosts=@llap0 or as extra configuration parameters for spark notebooks (ex : zeppelin) 3. Create a hiveWarhouse context import com.hortonworks.hwc.HiveWarehouseSession import com.hortonworks.hwc.HiveWarehouseSession._ val hive = HiveWarehouseSession.session(spark).build() //set a database hive.setDatabase("airline_ontime") //show table hive.showTables().show(100) 4. Query Data val flights_df = hive.executeQuery("SELECT * FROM flights WHERE year = 1989") flights_df.createOrReplaceTempView("flights_1998") +-----+----------+---------+-------+----------+-------+----------+-------------+---------+-------+-----------------+--------------+-------+--------+--------+------+----+--------+------+-------+---------+----------------+--------+------------+------------+--------+-------------+-----------------+----+ |month|dayofmonth|dayofweek|deptime|crsdeptime|arrtime|crsarrtime|uniquecarrier|flightnum|tailnum|actualelapsedtime|crselapsedtime|airtime|arrdelay|depdelay|origin|dest|distance|taxiin|taxiout|cancelled|cancellationcode|diverted|carrierdelay|weatherdelay|nasdelay|securitydelay|lateaircraftdelay|year| +-----+----------+---------+-------+----------+-------+----------+-------------+---------+-------+-----------------+--------------+-------+--------+--------+------+----+--------+------+-------+---------+----------------+--------+------------+------------+--------+-------------+-----------------+----+ | 12| 25| 1| 1415| 1415| 1547| 1552| US| 1478| NA| 92| 97| null| -5| 0| TPA| CLT| 508| null| null| 0| NA| 0| null| null| null| null| null|1989| 5. Write data Back to Hive ( in ACID Format ) hive.table("flights").filter("month = 01") .write .format(HiveWarehouseSession.HIVE_WAREHOUSE_CONNECTOR) .option("table", "flight_2019_01") .save() ... View more

Hi @Slim Thanks, yep that was it. Another little quirk identified with the help of Charles Bernard : - All names in the JSON object must be in lower case for them to be parsed - A corollary of this is that all columns names must also be in lower case ... View more

@Srikanth t The easiest approach is to use lateral views. https://cwiki.apache.org/confluence/display/Hive/LanguageManual+LateralView It allows you to split an array into multiple line. 1. Let's create an array from the items in your column "items" select key, split(items, ',') as valArray from test result +------+---------------------------------------+--+ | key | _c1 | +------+---------------------------------------+--+ | 22 | ["1001 abc"," 1002 pqr"," 1003 tuv"] | | 33 | ["1004 def"," 1005 xyz"] | +------+---------------------------------------+--+ 2. Now let's use lateral view to split these items into lines (using "trim" to clean up the space) select key, trim(uniqueVal) from( select key, split(items, ',') as valArray from test ) a lateral view explode(a.valArray) exploded as uniqueVal ; +------+-----------+--+ | key | _c1 | +------+-----------+--+ | 22 | 1001 abc | | 22 | 1002 pqr | | 22 | 1003 tuv | | 33 | 1004 def | | 33 | 1005 xyz | +------+-----------+--+ 3. Finally let's use split again to get separate values. select key, split(trim(uniqueVal), ' ')[0], split(trim(uniqueVal), ' ')[1] from( select key, split(items, ',') as valArray from test ) a lateral view explode(a.valArray) exploded as uniqueVal ; +------+-------+------+--+ | key | _c1 | _c2 | +------+-------+------+--+ | 22 | 1001 | abc | | 22 | 1002 | pqr | | 22 | 1003 | tuv | | 33 | 1004 | def | | 33 | 1005 | xyz | +------+-------+------+--+ Note : I used the following to create the table create table test ( key string, value string ) STORED AS ORC ; INSERT INTO test (key, value ) VALUES (22, '1001 abc, 1002 pqr, 1003 tuv'), (33, '1004 def, 1005 xyz'); ... View more

Hi @Michael Bronson Spark has a bunch of parameter to deal with job history rotation. In particular : spark.history.fs.cleaner.enabled true spark.history.fs.cleaner.maxAge  12h spark.history.fs.cleaner.interval 1h Source : https://spark.apache.org/docs/latest/monitoring.html In the example above : - Rotation is active - All Jobs > than 12 hours will be deleted - Deletion happens at 1 hour intervals Note these parameters need to be implemented on a environnement level ( not on a job level ). They are usually placed in spark-default file. Matthieu ... View more

Hello @Raju The thing to know when handling dataframes in Zeppelin is that a resultset is imported into the notebook when queried. That table is also read into memory when a particular notebook is accessed. To prevent a query for bringing back too much data and crashing the server (this can happen very quickly), each interpreter set a limit to the number of rows that are brought back to zeppelin ( default for most interpreters is ~1000 ). => This does not mean that queries of more than a thousand rows will not be executed, just that only the first 1000 rows are actually shown in Zeppelin. You can adjust the number of rows interperter by interpreter, look for "maxResult" properties Go to the interpreter configuration page (upper right hand corner) Ex : for SPARK zeppelin.spark.maxResult or for Livy zeppelin.livy.spark.sql.maxResult For the JDBC interperter ( there's always an exception to the rule 🙂 ) common.max_count When using Zeppelin's dataframe export to CSV feature you simply exports what has been pushed back to zeppelin. If the the max number of rows is set to a thousand, then you'll never have more than a thousand rows in your csv => The actual number of rows in your result set may be larger, it simply hasn't been fully read back into Zeppelin. This feature is great when working with small resultsets. It can however be deceiving as the results can be arbitrarily truncated when the max number of rows has been reached. If you're looking to export an entire table or a large subset, you should probably do it programmatically, for example by saving a table to a file system such as HDFS For Spark (2.0): dataframe.coalesce(1) .write .option("header", "true") .csv("/path/to/sample_file.csv") //Note the coalesce(1) => will bring all result to a single file otherwise you'll have 1 file per executor FYI Notebooks are simply JSON object organized by paragraph. Just open up any notebook to get a sense of the structure In HDP they are saved in : /usr/hdp/current/zeppelin-server/notebook/[id of notebook]/note.json ... View more

Correction for syntax using beeline : Formated data for referecence { "user": { "location": "", "id": 1514008171, "name": "Auzzie Jet", "screenname": "metalheadgrunge", "geoenabled": false }, "tweetmessage": "Anthrax - Black - Sunglasses hell yah\n http://t.co/qCNjba57Dm", "createddate": "2013-06-20T12:08:44", "geolocation": null } <br> From here on I refer to `user`, `tweetmessage`, `createddate`, `geolocation` => Level 1 fields `location`, `id`, `name`, `screenname`, `geoenabled` => Level 2 fields Corrected select query : SELECT t2.name, t1.tweetmessage FROM tweets t LATERAL VIEW json_tuple(t.tweet, 'user', 'tweetmessage' ) t1 as `user`, `tweetmessage` LATERAL VIEW json_tuple(t1.`user`, 'name', 'location') t2 as `name`, `location` where t2.`name`="Auzzie Jet" ; Other examples : Select level 1 : `user`, `tweetmessage`, `createddate`, `geolocation` SELECT t1.`user`, t1.tweetmessage, t1.createddate, t1.geolocation FROM tweets t LATERAL VIEW json_tuple(t.tweet, 'user', 'tweetmessage', 'createddate', 'geolocation' ) t1 as `user`, `tweetmessage`, `createddate`, `geolocation` ; Select level 1 and 2 => Flatten everything SELECT t2.location, t2.id, t2.name, t2.screenname, t2.geoenabled, t1.tweetmessage, t1.createddate, t1.geolocation FROM tweets t LATERAL VIEW json_tuple(t.tweet, 'user', 'tweetmessage', 'createddate', 'geolocation' ) t1 as `user`, `tweetmessage`, `createddate`, `geolocation` LATERAL VIEW json_tuple(t1.`user`, 'location', 'id', 'name', 'screenname', 'geoenabled' ) t2 as `location`, `id`, `name`, `screenname`, `geoenabled` ; ... View more

Hi @Sivasaravanakumar K The write function was implemented in 1.4.1... Try simply : df.saveAsTable("default.sample_07_new_schema") It will be saved as Parquet (default format for Spark) ... View more