核心算法代码分享如下：

#Flink连接HDFS上面的CSV文件 使用Flink_SQL分析完入表
## 启动hadoop
## cd /data/hadoop/sbin
## sh /data/hadoop/sbin/start-all.sh
## 启动hive
## cd /data/hive
## nohup hive --service metastore &
## nohup hive --service hiveserver2 &
import os
from pyflink.common import Row
from pyflink.table import (EnvironmentSettings, TableEnvironment, TableDescriptor, Schema,
                           DataTypes, FormatDescriptor)
from pyflink.table.expressions import lit, col
from pyflink.table.udf import udtf
import logging
import sys

logging.basicConfig(stream=sys.stdout, level=logging.INFO, format="%(message)s")

env_settings = EnvironmentSettings.in_streaming_mode()
table_env = TableEnvironment.create(env_settings)


jars = []
for file in os.listdir(os.path.abspath(os.path.dirname(__file__))):
    if file.endswith('.jar'):
        file_path = os.path.abspath(file)
        jars.append(file_path)

str_jars = ';'.join(['file:///' + jar for jar in jars])
table_env.get_config().get_configuration().set_string("pipeline.jars", str_jars)
# table_env.get_config().get_configuration().set_float('taskmanager.memory.network.fraction',0.8)
# table_env.get_config().get_configuration().set_string('taskmanager.memory.network.min','8gb')
# table_env.get_config().get_configuration().set_string('taskmanager.memory.network.max','16gb')
table_env.get_config().get_configuration().set_string('parallelism.default','1')
#先读取hadoop_hdfs上的CSV文件
table_env.execute_sql(
                """
                   create  table ods_earthquake(
                    `magnitude` double COMMENT '震级(M)',
                    `stime_long` string COMMENT '发震时刻(UTC+8)',
                    `latitude` string COMMENT '纬度(°)',
                    `longitude` string COMMENT '经度(°)',
                    `depth` bigint COMMENT '深度(千米)',
                    `addr`  string COMMENT '参考位置',
                    `stime_short` string COMMENT '年月日',
                    `data_type` string COMMENT '数据类型(用于区分中国外国)',
                    `long_province` string COMMENT 'echarts中国地图省份专用字段(外国是：无(外国))',
                    `stime_year` string COMMENT '年',
                    `stime_month` string COMMENT '月',
                    `stime_clock` string COMMENT '时刻中的时'
                    
                    ) WITH(
                           'connector' = 'filesystem',
                            'path' ='hdfs://bigdata:9000/earthquake/ods_earthquake/earthquake_hdfs.csv',
                            'format' = 'csv'
                )   
                """
                      )

#设置下沉到mysql的表
table_env.execute_sql(
                """
                create  table table06(
                    `addr` string primary key ,
                    `num` bigint 
                ) WITH(
                    'connector' = 'jdbc',
                    'url' = 'jdbc:mysql://bigdata:3306/hive_earthquake',
                    'table-name' = 'table06',
                    'username' = 'root',
                    'password' = '123456',
                    'driver' = 'com.mysql.jdbc.Driver'


                )   
                """
                      )


#数据分析并且导入
#result=table_env.sql_query("select * from ods_zymk limit 10 ")
table_env.execute_sql("""
                        insert into table06
                        select addr  , count(*) num
                        from ods_earthquake
                        GROUP BY addr 
                        order by num desc
                        limit 10;
                                                
                        


                  """).wait()
#print("表结构",result.get_schema())
#print("数据检查",result.to_pandas())