flink-table-common-.jar 官网下载

flink-table-common-.jar 是 Flink 中用于支持表格处理的共享库，可以用于访问和操作表格数据。想要下载这个库，可以通过去 Flink 官网进行下载。

在 Flink 官网上，我们可以找到一个名为 "下载" 或 "下载页面" 的菜单选项。点击进入下载页面后，会有列出不同版本的 Flink。根据自己的需求选择一个最新的版本。

在所选版本中，可以找到 "共享库" 或 "附加组件" 的选项。点击进入后，会有列出不同的库和组件。在列表中查找 "flink-table-common-.jar" 作为需要下载的共享库。

通常，下载链接会提供对应的二进制文件以及相关的文档和示例代码。确保选择正确的操作系统和对应的 Flink 版本后，点击下载按钮即可。

下载完成后，可以将 flink-table-common-.jar 文件导入到你的 Flink 项目中，并在代码中引入该库，即可使用其中的 API 进行表格数据的处理和操作。

总结起来，要下载 flink-table-common-.jar，可以通过 Flink 官网的下载页面找到对应的版本，并选择该库进行下载。下载完成后，将其导入到项目中，并按照文档和示例代码使用其中的 API 进行开发。

相关问题

Caused by: java.io.FileNotFoundException: JAR entry META-INF/services/org.apache.flink.table.factories.TableFactory not found in /opt/module-2.1.0/streampark_workspace/workspace/100013/streampark-flinkjob_QueryStoreResourceUDTF.jar

问题的原因是在提交任务时，指定的JAR文件中缺少了META-INF/services/org.apache.flink.table.factories.TableFactory这个文件。这个文件是Flink用来加载TableFactory的配置文件，缺少它会导致文件找不到的错误。解决这个问题的方法是确保在打包JAR文件时，将这个文件包含在内。你可以检查一下你的打包脚本或者构建工具的配置，确保这个文件被正确地包含在JAR文件中。
#### 引用[.reference_title]
- *1* [org.apache.flink.client.program.ProgramInvocationException: Could not retrieve the execution result.](https://blog.csdn.net/yiyayiyayi_ya/article/details/112242176)[target="_blank" data-report-click={"spm":"1018.2226.3001.9630","extra":{"utm_source":"vip_chatgpt_common_search_pc_result","utm_medium":"distribute.pc_search_result.none-task-cask-2~all~insert_cask~default-1-null.142^v91^insertT0,239^v3^insert_chatgpt"}} ] [.reference_item]
[ .reference_list ]

flink通过flink-connector-tidb-cdc使用streamapi方式连接tidb， 并展示相关代码

下面是使用 Flink Connector TiDB CDC 通过 Stream API 连接 TiDB 的示例代码：

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.tidb.JdbcConnectionOptions;
import org.apache.flink.streaming.connectors.tidb.TiDBOptions;
import org.apache.flink.streaming.connectors.tidb.TiDBSink;
import org.apache.flink.streaming.connectors.tidb.TiDBSource;
import org.apache.flink.streaming.connectors.tidb.TransactionIsolation;
import org.apache.flink.streaming.connectors.tidb.TiDBCatalog;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.api.bridge.java.internal.StreamTableEnvironmentImpl;
import org.apache.flink.types.Row;

import java.util.Properties;

public class TiDBStreamExample {

    public static void main(String[] args) throws Exception {

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

        // Define TiDB catalog
        TiDBCatalog catalog = new TiDBCatalog("tidb_catalog", "default_database",
                JdbcConnectionOptions.builder()
                        .withUrl("jdbc:mysql://tidb_ip:tidb_port/tidb_database_name?useSSL=false&useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC")
                        .withUsername("tidb_username")
                        .withPassword("tidb_password")
                        .build(),
                TiDBOptions.builder().withDatabaseUrl("jdbc:mysql://tidb_ip:tidb_port/tidb_database_name").build());
        tEnv.registerCatalog("tidb_catalog", catalog);
        tEnv.useCatalog("tidb_catalog");

        // Define TiDB source
        TiDBSource source = TiDBSource.builder()
                .setDatabaseName("tidb_database_name")
                .setTableName("tidb_table_name")
                .setOptions(TiDBOptions.builder()
                        .withDatabaseUrl("jdbc:mysql://tidb_ip:tidb_port/tidb_database_name")
                        .build())
                .setPrimaryKey("id")
                .setTransactionIsolation(TransactionIsolation.READ_COMMITTED)
                .build();

        // Create a data stream from TiDB source
        DataStream<Row> stream = env.addSource(source);

        // Define Flink Kafka producer
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "kafka_ip:kafka_port");
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<String>(
                "kafka_topic",
                new SimpleStringSchema(),
                props);

        // Map the data stream to a string stream and send it to Kafka
        DataStream<String> stringStream = stream.map(new MapFunction<Row, String>() {
            @Override
            public String map(Row row) throws Exception {
                return row.toString();
            }
        });
        stringStream.addSink(kafkaProducer);

        // Define Flink Kafka consumer
        FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(
                "kafka_topic",
                new SimpleStringSchema(),
                props);

        // Create a data stream from Kafka
        DataStream<String> kafkaStream = env.addSource(kafkaConsumer);

        // Convert the Kafka stream to a table and register it in the table environment
        tEnv.createTemporaryView("kafka_table", kafkaStream, "value");

        // Query the table and print the result to console
        tEnv.sqlQuery("SELECT * FROM kafka_table").execute().print();

        // Define TiDB sink
        TiDBSink sink = TiDBSink.builder()
                .setDatabaseName("tidb_database_name")
                .setTableName("tidb_table_name")
                .setOptions(TiDBOptions.builder()
                        .withDatabaseUrl("jdbc:mysql://tidb_ip:tidb_port/tidb_database_name")
                        .build())
                .setPrimaryKey("id")
                .build();

        // Convert the Kafka stream back to a data stream of rows and write it to TiDB
        DataStream<Row> rowStream = kafkaStream.map(new MapFunction<String, Row>() {
            @Override
            public Row map(String value) throws Exception {
                String[] fields = value.split(",");
                return Row.of(Integer.parseInt(fields[0]), fields[1], Double.parseDouble(fields[2]));
            }
        });
        rowStream.addSink(sink);

        env.execute("TiDB Stream Example");
    }
}

在上面的示例代码中，我们首先定义了一个 TiDBCatalog 对象，用于连接 TiDB 数据库。然后，我们使用 TiDBSource.builder() 方法定义了一个 TiDB 数据源，用于从 TiDB 中读取数据。接着，我们使用 env.addSource(source) 方法创建了一个 Flink 数据流。我们还定义了一个 Flink Kafka 生产者，用于将数据流发送到 Kafka。为了将数据流转换为字符串流，我们使用了 map() 方法。然后，我们将字符串流发送到 Kafka。接着，我们定义了一个 Flink Kafka 消费者，用于从 Kafka 中读取数据。我们还将 Kafka 数据流转换为表，并在表环境中注册它。最后，我们使用 TiDBSink.builder() 方法定义了一个 TiDB 汇聚器，用于将数据流写入 TiDB 中。

请注意，在上面的示例代码中，我们使用了 TiDBCatalog 和 TiDBSource 类来连接 TiDB 数据库。这些类需要 TiDB Connector JAR 包的支持。如果您没有安装该 JAR 包，请按照以下步骤安装：

1. 访问 TiDB Connector JAR 包的下载页面：https://github.com/pingcap/tidb/releases/tag/v4.0.8
2. 下载适用于您的操作系统的 JAR 包
3. 将 JAR 包添加到您的项目依赖中

最后，记得将代码中的 tidb_ip、tidb_port、tidb_database_name、tidb_table_name、tidb_username 和 tidb_password 替换为实际的值。同样，将代码中的 kafka_ip、kafka_port 和 kafka_topic 替换为实际的值。