hadoop - Hive Create Multi small files for each insert in HDFS -

-

following been achieved

  1. kafka producer pulling data twitter using spark streaming.
  2. kafka consumer ingesting data hive external table(on hdfs).

while working fine far. there 1 issue facing, while app insert data hive table, created small file each row data per file.

below code

// define topics read   val topic = "topic_twitter"   val groupid = "group-1"   val consumer = kafkaconsumer(topic, groupid, "localhost:2181")  //create sparkcontext   val sparkcontext = new sparkcontext("local[2]", "kafkaconsumer")  //create hivecontext     val hivecontext = new org.apache.spark.sql.hive.hivecontext(sparkcontext)    hivecontext.sql("create external table if not exists twitter_data (tweetid bigint, tweettext string, username string, tweettimestamp string,   userlang string)")   hivecontext.sql("create external table if not exists demo (foo string)")

hive demo table populated 1 single record. kafka consumer loop thru data topic ="topic_twitter" in process each row , populate in hive table 

val hivesql = "insert table twitter_data select stack( 1," +      tweetid        +","  +      tweettext      +"," +      username       +"," +     tweettimestamp +","  +     userlang + ") demo limit 1"  hivecontext.sql(hivesql)

below images hadoop environment. twitter_data, demo hie tables in hdfs

last 10 files created in hdfs enter image description here

as can see file size not more 200kb, there way merge these files in 1 file?

[take 2] ok, can't "stream" data hive. can add periodic compaction post-processing job...

  • create table 3 partitions e.g. (role='collecta'), (role='collectb'), (role='archive')
  • point spark inserts (role='activea')
  • at point, switch (role='activeb')

  • then dump every record have collected in "a" partition "archive", hoping hive default config job of limiting fragmentation

    insert table twitter_data partition (role='archive') select ... twitter_data role='activea' ; truncate table twitter_data partition (role='activea') ;

  • at point, switch "a" etc.

one last word: if hive still creates many files on each compaction job, try tweaking parameters in session, before insert e.g.

set hive.merge.mapfiles =true; set hive.merge.mapredfiles =true; set hive.merge.smallfiles.avgsize=1024000000;

Comments

Post a Comment

Popular posts from this blog

mysql - FireDac error 314 - but DLLs are in program directory -

-
i getting when trying access mysql database : [firedac][phys][mysql]-314. cannot load vendor library [libmysql.dll or libmysqlld.dll] this did not happen (unchanged) code, however, have upgraded windows 10 , had reinstall delphi xe8, system configuration matter. in order try solve problem, copied both of files c:\windows\sysytem32. when did not seem work, copied them \win32\debug, generated .exe resides. i imagine doing rather stupid, can't see what. the proper solution place driver file (eg., libmysql.dll ) in application's folder, or place installation location in fddrivers.ini file: [mysql] vendor=<folder>\libmysql.dll (recent versions of documentation seem use vendorlib instead of vendor in ini file.) see rad studio documentation topics configuring drivers (firedac) , connect mysql server (firedac) more information.
Read more

Log not being recorded for quickfix c++ Session -

-
i running fix initiator , succesfully connecting acceptor. problem is, no log being recorded. .cfg application written below. [default] connectiontype=initiator reconnectinterval=2 resetonlogon=y filestorepath=store filelogpath=logging starttime=00:00:00 endtime=00:00:00 usedatadictionary=y # standard config elements [session] # inherit connectiontype, reconnectinterval , sendercompid default beginstring=fix.4.4 sendercompid=init targetcompid=accept socketconnecthost=xxx socketconnectport=xxx heartbtint=30 datadictionary=fix44md.xml [session] beginstring=fix.4.4 sendercompid=init targetcompid=accept1 socketconnecthost=xxx socketconnectport=xxx heartbtint=30 datadictionary=fix44oms.xml the excerpt code initiates connection written below: std::string file = argv[ 1 ]; fix::sessionsettings settings( file ); application application; fix::filestorefactory storefactory( settings ); fix::screenlogfactory logfactory( settings ); fix::socketinitiator initi...
Read more

ios - NSFetchedResultsController not calling delegate methods on changes using NSPredicate with subquery -

-
i have nsfetchedresultscontroller created follows: let fetchrequest = nsfetchrequest(entityname: "friendship") fetchrequest.sortdescriptors = [ nssortdescriptor(key: "friendtocurrentuser.displayname", ascending: true) ] fetchrequest.predicate = nspredicate(format: "subquery(friendtocurrentuser.trackingsessions, $x, $x.dataisstale == false , $x.enddate == nil).@count != 0") let fetchedresultscontroller = nsfetchedresultscontroller(fetchrequest: fetchrequest, managedobjectcontext: coredatamanager.instance.mainmanagedobjectcontext, sectionnamekeypath: nil, cachename: nil) fetchedresultscontroller.delegate = self var error: nserror? if !fetchedresultscontroller.performfetch(&error) { println(error) } when initial fetch completes controller:didchangeobject:atindexpath:forchangetype: called, when object should added/removed due subquery method not called. if simple delegate methods aren't called when predicate contains subquery tha...
Read more