luigi操作hive表

关于luigi框架下查询hive表的操作

class JoinQuery(HiveQueryTask):date=luigi.DateParameter()def hiveconfs(self):jcs = {}jcs['mapred.job.name'] = "xxx_xxx_hive_daily_{}_username".format(format_date(self.date))jcs['mapred.job.queue.name'] = 'root.xxx.xxx'return jcsdef requires(self):return LogHiveFiles(date=self.date)def output(self):return luigi.hdfs.HdfsTarget(hdfs_targets.DailyExprImgHdfsFiles(date=self.date).path,format=luigi.hdfs.PlainDir)def query(self):query="""INSERT OVERWRITE DIRECTORY '{ot}'ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'SELECT * FROM {tb} WHERE date='{dt}' LIMIT 1""".format(ot=self.output().path,tb=self.input().table,dt=self.input().partition['date'])return queryclass recCountStats(luigi.Task):now = datetime.now()date = luigi.DateParameter(default=datetime(now.year, now.month, now.day) - timedelta(days=3))def requires(self):return JoinQuery(self.date)
if __name__ == '__main__':luigi.run(main_task_cls=recCountStats)

直接上的代码，luigi框架不多说了，可以看之前的luigi的文章。HiveQueryTask类是封装好的专门用于hive sql的类，就是把hive的query写在程序中，他帮你查。

1.JoinQuery类是直接继承自HiveQueryTask类的，处理Luigi本身就有的requires和output以外，区别于spark任务，不用写main()或run()方法，写query（）方法就好。

2.区别于luigi下执行spark任务，执行hive和执行mapreduce比较像，需要定义hiveconf方法，就是程序的配置。里面需要定义程序的名称和跑该程序的队列，切记，写了名称和队列 公司服务器才会接受运行该程序，要不然报错的。

3.关于查询的query语句，Luigi不捕捉查询任务的输出，要用INSERT DIRECTORY系列语句将查询结果持久化，并作为output输出。看代码！INSERT OVERWRITE DIRECTORY +table名  可以把结果写入hdfs表 ，用LOCAL DIRECTORY的话可以写在本地。

4.关于requires函数中的 返回的类，可参考下面这样定义，和hdfs的有一些区别，hdfs的只要一个地址就行了。hive的：

class LogHiveFiles(luigi.ExternalTask):date = luigi.DateParameter()def output(self):return luigi.hive.HivePartitionTarget(table='xxxs_xxx_log_xxx_daily',partition={'date': format_date(self.date)})

用的是HivePartitionTarget类。

转载于:https://www.cnblogs.com/qingjiaowoyc/p/7028217.html

本文来自互联网用户投稿，文章观点仅代表作者本人，不代表本站立场，不承担相关法律责任。如若转载，请注明出处。 如若内容造成侵权/违法违规/事实不符，请点击【内容举报】进行投诉反馈！