假设我们有这样的工作:
class MRjob(JobTask):
def output(self):
return ...
def requires(self):
return ...
def mapper(self, line):
# some line process
yield key, (...information, stored in hashable type...)
def reducer(self,key,values):
# some reduce logic... for example this
unique = set(values)
for elem in unique:
yield key, elem[0], elem[1]
在输出方法中我应该怎么做才能将数据插入现有的表分区(表格也以orc格式存储)?我想跳过将数据转换为orc的过程,因此我尝试了
return HivePartitionTarget(self.insert_table, database=self.database_name, partition=partition)
但这没用。我还发现luigi试图将输出传递给某个文件。使用HivePartitionTarget,luigi返回错误,如'object has no attribute write',所以我的假设是HivePartitionTarget只是不包含write方法。因此,我认为我做错了,应该使用另一种方法,但没有设法找到一个例子
关于如何在luigi
实现这一点,我不太了解。我可能建议用简单的方法以正常分隔格式(例如逗号分隔格式)编写luigi
脚本的输出。
在其上创建一个外部配置单元表:
CREATE EXTERNAL TABLE temp_table(
<col_name> <col_type>,
<col_name2> <col_type>
.......
.......
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ‘,’
LOCATION ‘ /hive/data/weatherext’;
使用简单的hive insert-into-select
查询将数据插入原始表(具有ORC格式数据)。
INSERT INTO TABLE target_table
PARTITION( xxx )
SELECT
COL_NAME1,
COL_NAME2
FROM temp_table;
您的目标表将具有ORC格式的数据,而hive将为您处理转换。