如何用Python在Hive路径文件中插入列并写入Kafka消息与系统日期

一、在Hive路径文件中插入两列（带分隔符） #

Hive存储路径对应HDFS目录，操作需保证文件格式与Hive表分隔符一致（常见如\t、,），以下两种方案可选：

方案1：直接修改HDFS文件（离线处理已有文件） #

需借助hdfs库连接HDFS，读取文件后每行追加指定列，再写回：

from hdfs import InsecureClient
import datetime

# 连接HDFS集群
client = InsecureClient('http://hdfs-nn:50070', user='hive')
# Hive表对应的HDFS文件路径
hive_file_path = '/user/hive/warehouse/your_db.your_table/data_file'

# 定义要插入的两列数据
sys_date = datetime.datetime.now().strftime('%Y-%m-%d')
source_flag = 'kafka_import'

# 读取并修改文件内容
with client.read(hive_file_path, encoding='utf-8') as f:
    lines = f.readlines()

modified_lines = [f"{line.strip()}\t{sys_date}\t{source_flag}\n" for line in lines]

# 覆盖写入原文件（生产环境建议先备份）
with client.write(hive_file_path, encoding='utf-8', overwrite=True) as f:
    f.writelines(modified_lines)

方案2：通过Hive SQL操作（规范且适配Hive表） #

先给Hive表新增列，再用SQL批量更新：

1. 执行ALTER TABLE添加列：

ALTER TABLE your_db.your_table ADD COLUMNS (sys_date STRING, source_flag STRING);

2. Python连接Hive执行更新：

from hive import Connection

conn = Connection(host="xxxx", port=xx, username="xx", password="xx", auth="CUSTOM")
cursor = conn.cursor()

# 批量更新新增列
update_sql = """
UPDATE your_db.your_table 
SET sys_date = current_date(), source_flag = 'kafka_import'
"""
cursor.execute(update_sql)
conn.commit()

cursor.close()
conn.close()

二、修改Kafka消费代码，追加系统日期到每条记录 #

引入datetime模块获取系统日期，将其与payload按分隔符拼接后写入文件，调整后的代码如下：

from kafka import TopicPartition
from hive import Connection
import datetime
import json  # 可选，若payload为JSON格式使用

topicPartition=[TopicPartition(mytopic, p) for p in range(0,mytopicpartitions)]
consumer.assign(topicPartition)
consumer.seek_to_beginning()

print("start hive connection")
conn = Connection(host="xxxx", port=xx, username="xx",password="xx",auth="CUSTOM")
print("end hive connection")

# 用with语句管理文件，自动处理关闭
with open("/location/kafkafeed.txt", "w") as f:
    for msg in consumer:
        payload = msg.value.decode('utf-8')
        # 获取系统日期，可自定义格式（如'%Y-%m-%d %H:%M:%S'包含时间）
        sys_date = datetime.datetime.now().strftime('%Y-%m-%d')
        
        # 方式1：直接用分隔符拼接payload和日期（适合非JSON格式的payload）
        f.write(f"{payload}\t{sys_date}\n")
        
        # 方式2：若payload是JSON，将日期作为字段插入后序列化（推荐，避免解析问题）
        # payload_dict = json.loads(payload)
        # payload_dict['sys_date'] = sys_date
        # f.write(f"{json.dumps(payload_dict)}\n")
        
        print("writing into a file")

关键注意点 #

• 确保文件写入的分隔符与后续Hive表的字段分隔符一致，否则Hive读取会出现数据错位
• 写入Hive路径的文件时，需保证文件权限符合Hive要求（通常属主为hive用户），否则Hive无法加载数据

内容的提问来源于stack exchange，提问作者dev_code