flink hive 2 msyql

您可以使用Flink的Hive集成来将Hive表映射到MySQL表。具体步骤如下：

1. 在Hive中创建表，并将其存储在MySQL中。

  CREATE EXTERNAL TABLE mysql_table (
    col1 STRING,
    col2 INT,
    col3 DOUBLE
  )
  STORED BY 'org.apache.hadoop.hive.mysql.MySQLStorageHandler'
  WITH SERDEPROPERTIES (
    'javax.jdo.option.ConnectionURL'='jdbc:mysql://localhost:3306/mydb',
    'javax.jdo.option.ConnectionUserName'='user',
    'javax.jdo.option.ConnectionPassword'='password',
    'mysql.table.schema'='mydb',
    'mysql.table.name'='mysql_table'
  );

2. 在Flink中设置Hive配置。

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
  StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

  Map<String, String> hiveConf = new HashMap<>();
  hiveConf.put("hive.metastore.uris", "thrift://localhost:9083");
  hiveConf.put("hive.exec.dynamic.partition.mode", "nonstrict");
  hiveConf.put("hive.default.fileformat", "TextFile");
  hiveConf.put("javax.jdo.option.ConnectionURL", "jdbc:mysql://localhost:3306/mydb");
  hiveConf.put("javax.jdo.option.ConnectionUserName", "user");
  hiveConf.put("javax.jdo.option.ConnectionPassword", "password");

  tableEnv.getConfig().getConfiguration().setStrings(HiveCatalogFactoryOptions.HIVE_SITE_XML.key(), "/path/to/hive-site.xml");
  tableEnv.registerCatalog("myhive", new HiveCatalog("myhive", "default", hiveConf));
  tableEnv.useCatalog("myhive");

3. 现在，您可以在Flink中使用Hive表。

  Table result = tableEnv.sqlQuery("SELECT * FROM mysql_table");
  DataStream<Row> stream = tableEnv.toAppendStream(result, Row.class);
  stream.print();

请注意，您需要在Flink的classpath中包含MySQL驱动程序。