如何使用 Python（不使用 PySpark）将 pandas 数据框插入现有的 Hive 外部表？

我正在创建一个到 Hive 的连接字符串，并在该连接上的 Hive 表上运行一些 SELECT 查询。

对检索到的数据执行一些转换后，我正在创建一个数据框df_student_credits看起来如下

NAME_STUDENT_INITIAL    CREDITS_INITIAL     NAME_STUDENT_FINAL  CREDITS_FINAL   LOAD_DATE
John                    23                  John                25              21/03/2017
Alan                    19                  Alan                19              17/06/2018
Will                    24                  Will                26              02/08/2019
Lily                    25                  Lily                25              22/01/2019

现在，我想将此数据帧插入到我使用以下命令创建的 Hive 外部表中：

CREATE EXTERNAL TABLE IF NOT EXISTS school_db.student_credits
(
NAME_STUDENT_INITIAL STRING,
CREDITS_INITIAL STRING,
NAME_STUDENT_FINAL STRING,
CREDITS_FINAL STRING,
LOAD_DATE STRING
)
LOCATION '/user/gradebook/student_credits/';

我希望每次运行此脚本并生成数据帧时将插入内容附加到表中，而不是覆盖现有的表数据。

我看到的几乎每一篇文章都展示了使用 PySpark 执行此操作的方法。但我无法使用 PySpark，我必须使用与触发 SELECT 查询相同的 python 脚本来实现此目的。

我是 Hive 的新手，也是 Python 的新手。有人可以帮我解决这个问题吗？

看来您正在尝试从 Hive 表读入 pandas 数据帧并进行一些转换并将其保存回某个 Hive 外部表。请参考下面的代码作为示例。在这里，我已从 Hive 表读取到 pandas 数据框中，并向其中添加了一些日期列。后来我使用 subprocess 模块来执行我的 shell，它将数据加载到按某些日期列分区的 Hive 表中。

from pyhive import hive
import pandas as pd
import sqlalchemy
from sqlalchemy.engine import create_engine
import datetime
from subprocess import PIPE, Popen
import subprocess
import sys

conn = hive.Connection(host="yourhost.com", port=10000, username="vikct001")
cursor = conn.cursor()

query="select user_id,country from test_dev_db.test_data"

start_time= datetime.datetime.now()

output_file='/home/vikct001/user/vikrant/python/test_data.csv'

data=pd.read_sql(query,conn)
data['current_date'] = pd.datetime.today().strftime("%Y-%m-%d")
print(data)

data.to_csv(output_file, sep='|', encoding='utf-8',index=None)

hivequery=""" hive --hivevar loaded_date=$(date +"%Y-%m-%d") hive -e 'LOAD DATA LOCAL INPATH "/home/vikct001/user/vikrant/python/test_data.csv" INTO TABLE test_dev_db.test_data_external PARTITION (loaded_date="${hivevar:loaded_date}")';"""

def save_to_hdfs(output_file):
        print("I am here")
        p=subprocess.Popen(hivequery,shell=True,stderr=subprocess.PIPE)
        stdout,stderr = p.communicate()
        if p.returncode != 0:
            print stderr
            sys.exit(1)


save_to_hdfs(output_file)
end_time=datetime.datetime.now()

print 'processing ends', (start_time-end_time).seconds/60.0,' minutes'

表说明：

hive (test_dev_db)> desc test_dev_db.test_data_external;
OK
id                      int
country                 string
input_date              date
loaded_date             string

# Partition Information
# col_name              data_type               comment

loaded_date             string

您可以看到数据已加载并创建了具有当前日期的分区。

hive (test_dev_db)> show partitions test_dev_db.test_data_external;
OK
loaded_date=2019-08-21


hive (test_dev_db)> select * from test_dev_db.test_data_external;
OK
1       India   2019-08-21      2019-08-21
2       Ukraine 2019-08-21      2019-08-21
1       India   2019-08-21      2019-08-21
2       Ukraine 2019-08-21      2019-08-21
1       India   2019-08-21      2019-08-21
2       Ukraine 2019-08-21      2019-08-21
1       India   2019-08-21      2019-08-21