17. PyFlink 集成 hive catalog模式读写数据

conda create --name pyflink python=3.8

conda activate pyflink

python -m pip install apache-flink==1.15.2

cp opt/software/flink-1.15.2/lib/* /opt/software/miniconda3/envs/pyflink/lib/python3.8/site-packages/pyflink/lib

from pyflink.table import *
from pyflink.table.catalog import HiveCatalog, CatalogDatabase, ObjectPath, CatalogBaseTable
catalog = HiveCatalog("myhive","caijing01", "/opt/software/hive/conf")
t_env = TableEnvironment.create(environment_settings=EnvironmentSettings.in_batch_mode())
t_env.register_catalog("myhive", catalog)
t_env.execute_sql("select * from myhive.caijing01.score5").print()
t_env.execute_sql("create table myhive.caijing01.score8 like  myhive.caijing01.score5")
t_env.execute_sql("insert into myhive.caijing01.score8 select * from  myhive.caijing01.score5")

from pyflink.table import *
from pyflink.table.catalog import HiveCatalog
settings = EnvironmentSettings.in_batch_mode()
t_env = TableEnvironment.create(settings)
source_ddl = """CREATE CATALOG myhive WITH ( 'type' = 'hive','default-database'='caijing01','hive-conf-dir' = '/opt/software/hive/conf')"""
t_env.execute_sql(source_ddl)
sql1= """select * from myhive.caijing01.score5"""
t_env.execute_sql(sql1).print()

 opt/software/flink-1.15.2/bin/flink run -pyexec opt/software/miniconda3/envs/pyflink/bin/python -py opt/software/flink-1.15.2/code/catalog.py