Flink 中如何将 DataStream 或 DataSet 转换为 Table？

推荐答案

在 Apache Flink 中，可以通过 StreamTableEnvironment 或 BatchTableEnvironment 将 DataStream 或 DataSet 转换为 Table。以下是具体的代码示例：

1. 将 DataStream 转换为 Table

-- -------------------- ---- -------
------ ----------------------------------------------------------------------
------ -----------------------------------------------
------ ---------------------------------
------ --------------------------------------------------------------

------ ----- ------------------------ -
    ------ ------ ---- ------------- ----- -
        -- -- --------------------------
        -------------------------- --- - -----------------------------------------------------
        
        -- -- ----------------------
        ------------------- -------- - ------------------------------------------------------------------------------
        ---------------------- -------- - ---------------------------------- ----------
        
        -- ---- ----------
        ------------------ ---------- - ------------------------- ------ --------
        
        -- - ---------- --- -----
        ----- ----- - ----------------------------------- --------
        
        -- -- -----
        --------------------
    -
-

2. 将 DataSet 转换为 Table

-- -------------------- ---- -------
------ -----------------------------------------------
------ -----------------------------------------------
------ ---------------------------------
------ -------------------------------------------------------------

------ ----- --------------------- -
    ------ ------ ---- ------------- ----- -
        -- -- --------------------
        -------------------- --- - -----------------------------------------------
        
        -- -- ---------------------
        ------------------- -------- - --------------------------------------------------------------------------
        --------------------- -------- - --------------------------------- ----------
        
        -- ---- -------
        --------------- ------- - ------------------------- ------ --------
        
        -- - ------- --- -----
        ----- ----- - ----------------------------- --------
        
        -- -- -----
        --------------------
    -
-

本题详细解读

1. 环境设置

在 Flink 中，StreamTableEnvironment 和 BatchTableEnvironment 是用于处理 Table API 和 SQL 的主要入口。StreamTableEnvironment 用于流处理，而 BatchTableEnvironment 用于批处理。

• StreamTableEnvironment 通常与 StreamExecutionEnvironment 一起使用。
• BatchTableEnvironment 通常与 ExecutionEnvironment 一起使用。

2. 转换方法

• fromDataStream: 用于将 DataStream 转换为 Table。可以指定字段名称，以便在 Table API 或 SQL 中使用。
• fromDataSet: 用于将 DataSet 转换为 Table。同样可以指定字段名称。

3. 字段名称

在转换过程中，可以为 DataStream 或 DataSet 中的元素指定字段名称。这些字段名称将在 Table API 或 SQL 中使用。例如，tableEnv.fromDataStream(dataStream, "name") 中的 "name" 就是字段名称。

4. 打印 Schema

转换后的 Table 可以通过 printSchema() 方法打印其结构，以便查看字段名称和类型。

5. 注意事项

• 在使用 StreamTableEnvironment 时，确保 EnvironmentSettings 设置为流模式（inStreamingMode()）。
• 在使用 BatchTableEnvironment 时，确保 EnvironmentSettings 设置为批模式（inBatchMode()）。
• Flink 的 Table API 和 SQL 依赖于 Blink Planner 或 Old Planner，因此在创建 TableEnvironment 时需要指定使用的 Planner。