标签：总结 city 入门 age tbl sql col pyFlink name

一　整体流程

1. 初始化pyFlink执行环境 2. 加载数据集 3. 执行数据分析 4. 导出分析结果   二 初始化执行环境

2.1 初始化

参考代码如下 from pyflink.table import EnvironmentSettings, StreamTableEnvironment es = EnvironmentSettings.new_instance().in_batch_mode().build() tv = StreamTableEnvironment.create(environment_settings=es)

2.2 其它

待补充其它初始化方法，如流处理等

三 加载数据集

3.1 基于变量

参考代码如下 data = [['T1', 34, 'XY'],['T2', 34, 'NY'],['T3', 33, 'XX'],['T4', 33, 'JZ'],['T5', 33, 'SZ'],['T6', 33, 'PDS'],['T7', 32, 'XC'],['T8', 32, 'NY']] tbl = tv.from_elements(data, ['name','age','city'],['STRING','INT','STRING']) tv.create_temporary_view('itable', tbl)  # 注册为flinksql能访问的对向

3.2 基于pandas.DataFrame

dfa = pd.DataFrame(data, columns='name age city'.split()) tbl = tv.from_pandas(dfa) 

3.3 基于csv

csv_path = 'iexample.csv' csv_schema = 'name string, age int, city string' csv_sql = F"create table iTable({csv_schema}) with ('connector' = 'filesystem', 'path' = '{csv_path}', 'format' = 'csv'))" tv.execute_sql(csv_sql) tbl = tv.from_path('iTable') 经验1 包含表头的csv会报错 经验2 csv_sql 后面的with 中的引号必须是 单引号 双引号会报错 经验3 不要创建重名的表，会报错 此条有待进一步确认  

3.4 连接postgresql

from pyflink.table import EnvironmentSettings, StreamTableEnvironment es = EnvironmentSettings.new_instance().in_batch_mode().build() tv = StreamTableEnvironment.create(environment_settings=es)   pg_schema = 'name STRING, age INT, city string' dsn = F'jdbc:postgresql://{host}:{port}/{database}' pg_sql = F"create table pg_table ({pg_schema})with ('connector'='jdbc','url'='{dsn}','table-name'='{table_name}','driver'='org.postgresql.Driver','username'='{user}','password'='{pwd}')" print(pg_sql) tv.execute_sql(pg_sql) tbl = tv.from_path('pg_table') tbl.limit(5).execute().print() 经验4 需要下载flink-connector-jdbc-*.jar文件 和 postgresql-*.jar文件 对应目录是pyflink安装目录的 lib文件夹下 经验5 连接依赖文件与数据库版本也有关系

四 数据处理

4.1 简单处理

1) select

from pyflink.table.expressions import col, call tt = tbl.select(col("city")) tt.limit(3).execute().print()

2)group_by

tbl.group_by(col('city')).select(col('city'),call("count", col('city')).alias('cnt')).execute().print() tv.register_table('itable', tbl) tv.sql_query('select city, count(*)cnt from itable group by city').execute().print()

3)order_by

tbl.order_by(col('age').desc).execute().print()

4)buildin function

tbl.select(call('avg',col('age')).alias('age_avg')).execute().print() tbl.select(call('sum',col('age')).alias('age_sum')).execute().print()

5)normalized

@udf(result_type='Row<_name string, _age float>', func_type='pandas')
def inormal(data: pd.DataFrame) -> pd.DataFrame:
    data['_age'] = (data['age']-data['age'].mean())/data['age'].std()
    return data[['name', '_age']]
tbl.map(inormal).execute().print()

6）map & udf

@udf(result_type='Row<_name string, province string>', func_type='pandas')
def itown(data: pd.DataFrame) -> pd.DataFrame:
    data['province'] = data.city.apply(lambda x: dct.get(x))
    return data[['name', 'province']]
tbl.map(itown).execute().print()

4.2 其它处理

待补充

五 输出

5.1 print

tbl.map(itown).execute().print()

5.2 CSV

# 定义输出 CSV 文件的 schema
sink_schema = "name STRING, age int, _age float, city string, province string"
# 定义输出 CSV 文件的目录
sink_path = "tmpfile"
# 注册输出表
tv.execute_sql(f"""
    CREATE TABLE sink_table (
        {sink_schema}
    ) WITH (
        'connector' = 'filesystem',
        'path' = '{sink_path}',
        'format' = 'csv'
    )
""")

_age = tbl.map(inormal)
_town = tbl.map(itown)
t1 = tbl.join(_age).where(col('name')==col('_name')).select(col('name'), col('age'), col('_age'), col('city'))
t2 = t1.join(_town).where(col('name')==col('_name')).select(col('name'), col('age'), col('_age'), col('city'), col('province'))
tv.create_temporary_view('jtable', t2)  # old api tv.register_table

# 执行查询并将结果写入输出表
sql = 'INSERT INTO sink_table(name, age, _age, city, province) SELECT name, age, _age, city, province FROM jtable'
tv.execute_sql(sql).wait()

 

经验6 csv输出路径只能指定目录 不能指定名称

经验7 数据库输出需要提前创建供写入的表

5.3 POSTGRESQL

pg_schema = "name STRING, age int, _age float, city string, province string"
dsn = F'jdbc:postgresql://{host}:{port}/{database}'
pg_sql = F"create table pg_table ({pg_schema})with ('connector'='jdbc','url'='{dsn}','table-name'='{table_name}','driver'='org.postgresql.Driver','username'='{user}','password'='{pwd}')"
tv.execute_sql(pg_sql)  # create

sql = 'INSERT INTO pg_table(name, age, _age, city, province) SELECT name, age, _age, city, province FROM jtable'
tv.execute_sql(sql).wait() 

标签：总结,city,入门,age,tbl,sql,col,pyFlink,name
From： https://www.cnblogs.com/ddzhen/p/18188654