sqoop与hive的使用

sqoop的使用•Sqoop是一个用来将Hadoop和关系型数据库中的数据相互转移的工具，可以将一个关系型数据库（例如:MySQL,Oracle,Postgres等,Sqoop支持任何一款支持JDBC规范的数据库）中的数据导进到Hadoop的HDFS中，也可以将HDFS的数据导进到关系型数据库中。Sqoop的使用•sqoopimport•--connectjdbc:oracle:thin:@172.16.2.30:1522:orcl•--usernameTEST--passwordTEST•--tableuniorder_order_info•--columnsORDER_NO,ORDER_STATUS,CHANNEL_NO,SECOND_CHANNEL_NO,THIRD_CHANNEL_NO,SITE_CODE,SHOPPING_TYPE,PAY_TYPE,TIME_STAMP,PAY_AMT,PAY_CURRENY,EXCHANGE_RMB,POINT_AMT,CASH_AMT,CASH_PAY_CURRENY,CASH_EXCHANGE_RMB,COUPON_AMT,COUPON_PAY_UNIT,RATE,PROMO_TYPE,PROMO_ID,CONTACT_NAME,CONTACT_MOBILE,USER_FFP_NO,CONTACT_EMAIL,REQUEST_IP,REMARK,IS_VALID,CRT_ID,CRT_DT,OPT_ID,OPT_DT,SUB_PROMO_TYPE,CANCEL_DATE,VIP_CODE,CALL_CONFIRM_STATUS,TRAVEL_ORDER_NO•--hive-import•--hive-tableUNIORDER_ORDER_INFO•--hive-partition-keyds--hive-partition-value20131201•--m8--fields-terminated-by'\t'--split-byORDER_NOSHXSqoop•苏研院开发的sqoop导入程序，通过修改配置文件，实现数据的导入。hive的使用Hive是建立在Hadoop上的数据仓库基础构架。它提供了一系列的工具，可以用来进行数据提取转化加载（ETL），这是一种可以存储、查询和分析存储在Hadoop中的大规模数据的机制。Hive定义了简单的类SQL查询语言，称为HQL，它允许熟悉SQL的用户查询数据。SQLMapReduceHive基本数据类型IntegersTINYINT-1byteSMALLINT-2byteINT-4byteBIGINT-8byteBooleantypeBOOLEAN-TRUE/FALSEFloatingpointnumbersFLOAT–单精度4byteDOUBLE–双精度8byteStringtypeSTRINGHiveQL复杂数据类型Structs:{a,’1’}Maps(key-valuetuples):'数学':80,'语文':89,'英语':95Arrays(indexablelists):[‘1’,‘2’,‘3’]字段类型必须相同托管表createtablestudent(idint,namestring)rowformatdelimitedfieldsterminatedby‘\t’;///指明分割符,默认是ctrl+a外部表createexternaltablestudentex(idint,namestring)rowformatdelimitedfieldsterminateby‘\t’location‘/user/hadoop/external_table’;不同点:load数据时,mangedtable会将数据移动到仓库目录,drop数据时会将表元数据和数据删除,而externaltable在load数据时不会移动数据.建表语句ALTERTABLEstudentADDCOLUMNS(new_colINT);为student表加入新列ALTERTABLEeventsRENAMETOnewName;修改表名表的修改loaddatainpath‘/user/hadoop/data/studentsource.txt’overwriteintotablestudent;Hive不支持一条一条的用insert语句进行插入操作，也不支持update的操作。数据是以load的方式，加载到建立好的表中。数据一旦导入，则不可修改。要么drop掉整个表，要么建立新的表，导入新的数据。overwrite关键字代表了先删除表目录里的文件,再将inpath后面的文件拷到表目录里,而不加的话,仅仅做将inpath后面的文件拷到表目录里INSERTOVERWRITEDIRECTORY'/tmp/hdfs_out'SELECTa.*FROMstudentaWHEREa.id5;将查询结果直接输出至HDFSinsertoverwriteLocalDIRECTORY'/tmp/test'select*fromuserwhereid=2;将查询结果直接输出至本地读取(插入)数据createtablelogs(tsbigint,linestring)partitionedby(dtstring,countrystring)rowformatfieldsterminatedby‘\t’导入数据loaddatainpath‘’intotablelogspartition(dt=’2013-09-10’,country=’GB’);insertoverwritetablelogslogspartition(dt=’2013-09-10’,country=’GB’)select*fromlog2;分区createtablebucketed_users(idint,namestring)clusteredby(id)sortedby(idasc)into4bucketsrowformatdelimitedfieldsterminatedby‘\t’;将普通表向桶表插入insertintooverwritetablebuc_userselect*fromusers;上面四个00000_0-4文件保存了所有记录分桶分桶抽样对桶中的数据进行采样：select*frombucketed_usertablesample(bucket1outof4onname);桶的个数从1开始计数，前面的查询从4个桶中的第一个桶获取数据。其实就是四分之一。内连接selectsales.*,things.*fromsalesjointhingson(sales.id=things.id);外连接selectsales.*,things.*fromsalesleftouterjointhingson(sales.id=things.id);半连接select*fromthingswherethings.idin(selectidfromsales);select*fromthingsleftsemijoinsaleson(sales.id=things.id);连接map连接select/*+mapjoin(things)*/sales.*,things.*fromsalesjointhingson(sales.id=things.id);解决数据倾斜问题，MAPJION会把小表全部读入内存中，在map阶段直接拿另外一个表的数据和内存中表数据做匹配，由于在map是进行了join操作，省去了reduce运行的效率也会高很多。hive自定义函数1、UDF操作单个数据行，产生单个数据行输出。2、UDAF(用户定义聚合函数)操作多个数据行，产生一个数据行输出。（如count、max等类型函数）UDF使用过程第一步：继承UDF或者UDAF，实现特定的方法。第二步：将写好的类打包为jar。如hive.jar.第三步：进入到Hive环境中，利用addjaraddjarhdfs://namenode:9000/user/hadoop/hive.jar.注册该jar文件第四步：为该类起一个别名，createtemporaryfunctiontestas'com.travelsky.hive';这里注意UDF只是为这个Hive会话临时定义的。第五步：在select中使用test();编写UDF新建java项目引入两个jar包hive-exec-0.10.0.jarhadoop-core-*.jar编写UDF类packagecom.tsinghua.udf;importorg.apache.hadoop.hive.ql.exec.UDF;importorg.apache.hadoop.io.Text;publicclassTestUDFextendsUDF{privateTextresult=newText();publicTextevaluate(Textstr){if(str==null){returnnull;}result.set(testUDF);returnresult;}}HiveUDF导出为jar包:myudf.jar进入hive命令行在hive中注册这个文件addjarhdfs://namenode:9000/user/hadoop/myudf.jar;为函数起个名字createtemporaryfunctiontestas‘包名.类名’;测试selecttest(‘abc’)fromuser;输出结果:testUDFtestUDF….HiveUDAFpackagecom.udaf;importorg.apache.hadoop.hive.ql.exec.UDAF;importorg.apache.hadoop.hive.ql.exec.UDAFEvaluator;importorg.apache.hadoop.io.IntWritable;publicclassMaximumextendsUDAF{publicstaticclassMaximumIntUDAFEvaluatorimplementsUDAFEvaluator{//最终结果privateIntWritableresult;//负责初始化计算函数并设置它的内部状态，result是存放最终结果的publicvoidinit(){//TODOAuto-generatedmethodstubresult=null;}//每次对一个新值进行聚集计算都会调用iterate方法publicbooleaniterate(IntWritablevalue){if(value==null)returnfalse;if(result==null)result=newIntWritable(value.get());elseresult.set(Math.max(result.get(),value.get()));returntrue;}//Hive需要部分聚集结果的时候会调用该方法publicIntWritableterminatePartial(){//TODOAuto-generatedmethodstubreturnresult;}//合并两个部分聚集值会调用这个方法publicbooleanmerge(IntWritableother){//TODOAuto-generatedmethodstubreturniterate(other);}publicIntWritableterminate(){returnresult;}}}UDAF的工作原理init()：主要是负责初始化计算函数并且重设其内部状态，一般就是重设其内部字段。iterate()：每一次对一个新值进行聚集计算时候都会调用该方法，计算函数会根据聚集计算结果更新内部状态。当输入值合法或者正确计算了，则就返回true。terminatePartial()：Hi