FlinkSQL split_index 的用法

以你之姓@ 2021-09-25 07:00 1067阅读 0赞

1 添加

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-table-planner-blink_${flink.scala.version}</artifactId>
        <version>${flink.version}</version>
    </dependency>

不然报错：

Exception in thread "main" org.apache.flink.table.api.TableException: Could not instantiate the executor. Make sure a planner module is on the classpath  
    at org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl$.lookupExecutor(StreamTableEnvironmentImpl.scala:330)  
    at org.apache.flink.table.api.bridge.scala.internal.StreamTableEnvironmentImpl$.create(StreamTableEnvironmentImpl.scala:285)  
    at org.apache.flink.table.api.bridge.scala.StreamTableEnvironment$.create(StreamTableEnvironment.scala:462)  
    at org.apache.flink.table.api.bridge.scala.StreamTableEnvironment$.create(StreamTableEnvironment.scala:432)  
    at TableTest.Example$.main(Example.scala:33)  
    at TableTest.Example.main(Example.scala)  
Caused by: org.apache.flink.table.api.NoMatchingTableFactoryException: Could not find a suitable table factory for 'org.apache.flink.table.delegation.ExecutorFactory' in  
the classpath.

[https://www.codenong.com/cs109776647/][https_www.codenong.com_cs109776647]

EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build(); //用useOldPlanner会报错

<table> 
 <tbody> 
  <tr> 
   <td> <pre><code>SPLIT_INDEX(string1, string2, integer1)</code></pre> </td> 
   <td> <p>Splits&nbsp;<em>string1</em>&nbsp;by the delimiter&nbsp;<em>string2</em>, returns the&nbsp;<em>integer</em>th (zero-based) string of the split strings. Returns NULL if&nbsp;<em>integer</em>&nbsp;is negative. Returns NULL if any of arguments is NULL.</p> <p><strong>Only supported in blink planner.</strong></p> </td> 
  </tr> 
 </tbody> 
</table>

[https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/functions.html][https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_table_functions.html]

package SQL_split_index;
    
    import org.apache.flink.streaming.api.datastream.DataStreamSource;
    import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    import org.apache.flink.streaming.api.functions.source.SourceFunction;
    import org.apache.flink.table.api.EnvironmentSettings;
    import org.apache.flink.table.api.Table;
    import org.apache.flink.table.api.java.StreamTableEnvironment;
    import org.apache.flink.types.Row;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Date;
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @Author you guess
     * @Date 2021/4/12 16:36
     * @Version 1.0
     * @Desc
     */
    public class DataStreamSql1_Split_Index {
        private static final Logger LOG = LoggerFactory.getLogger(DataStreamSql1_Split_Index.class);
        private static final String[] TYPE = {"a_苹果", "b_梨", "c_西瓜", "d_葡萄", "e_火龙果"};
    
        public static void main(String[] args) throws Exception {
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamTableEnvironment stEnv = StreamTableEnvironment.create(env, envSettings);
    
            //添加自定义数据源,每秒发出一笔订单信息{商品名称,商品数量}
            DataStreamSource<SQL.DataStreamSql1_Join.Order> orderSourceA = env.addSource(new SourceFunction<SQL.DataStreamSql1_Join.Order>() {
                private volatile boolean isRunning = true;
                private final Random random = new Random();
    
                @Override
                public void run(SourceContext<SQL.DataStreamSql1_Join.Order> ctx) throws Exception {
                    while (isRunning) {
                        TimeUnit.SECONDS.sleep(1);
                        SQL.DataStreamSql1_Join.Order order = new SQL.DataStreamSql1_Join.Order(TYPE[random.nextInt(TYPE.length)], Long.valueOf(random.nextInt(10)));
                        System.out.println(new Date() + ",orderSourceA提交元素:" + order);
                        ctx.collect(order);
                    }
                }
    
                @Override
                public void cancel() {
                    isRunning = false;
                }
    
            }, "order-infoA");
    
    
            stEnv.registerDataStream("tableA", orderSourceA);
    
            Table result = stEnv.sqlQuery("SELECT A.name,A.qtty,split_index(name,'_',0) as name0,split_index(name,'_',1) as name1  from tableA A");
            stEnv.toRetractStream(result, Row.class).print();//这里要用Row类型
            env.execute("Flink Streaming Java API Skeleton");
    
        }
    
    }

输出：

Mon Apr 12 17:05:10 CST 2021,orderSourceA提交元素:Order\{name='a\_苹果', qtty=2\}  
5> (true,a\_苹果,2,a,苹果)  
Mon Apr 12 17:05:11 CST 2021,orderSourceA提交元素:Order\{name='e\_火龙果', qtty=5\}  
6> (true,e\_火龙果,5,e,火龙果)  
Mon Apr 12 17:05:12 CST 2021,orderSourceA提交元素:Order\{name='a\_苹果', qtty=8\}  
7> (true,a\_苹果,8,a,苹果)  
Mon Apr 12 17:05:13 CST 2021,orderSourceA提交元素:Order\{name='c\_西瓜', qtty=4\}  
8> (true,c\_西瓜,4,c,西瓜)  
Mon Apr 12 17:05:14 CST 2021,orderSourceA提交元素:Order\{name='d\_葡萄', qtty=8\}  
1> (true,d\_葡萄,8,d,葡萄)

[https_www.codenong.com_cs109776647]: https://www.codenong.com/cs109776647/
[https_ci.apache.org_projects_flink_flink-docs-release-1.9_dev_table_functions.html]: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/functions.html