pom.xml文件引入

    2. <?xml version="1.0" encoding="UTF-8"?>
    3. <project xmlns="http://maven.apache.org/POM/4.0.0"
    4.          xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    5.          xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    6.     <modelVersion>4.0.0</modelVersion>
    8.     <groupId>org.example</groupId>
    9.     <artifactId>Spark_Test</artifactId>
    10.     <version>1.0-SNAPSHOT</version>
    11.     <build>
    12.         <plugins>
    13.             <plugin>
    14.                 <groupId>org.apache.maven.plugins</groupId>
    15.                 <artifactId>maven-compiler-plugin</artifactId>
    16.                 <configuration>
    17.                     <source>8</source>
    18.                     <target>8</target>
    19.                 </configuration>
    20.             </plugin>
    21.         </plugins>
    22.     </build>
    25.     <dependencies>
    27.         <dependency>
    28.             <groupId>ru.yandex.clickhouse</groupId>
    29.             <artifactId>clickhouse-jdbc</artifactId>
    30.             <version>0.1.40</version>
    31. <!--            <scope>test</scope>-->
    32.         </dependency>
    35. <!--        <dependency>-->
    36. <!--            <groupId>org.apache.flink</groupId>-->
    37. <!--            <artifactId>flink-clients_2.11</artifactId>-->
    38. <!--            <version>1.3.2</version>-->
    39. <!--        </dependency>-->
    41. <!--        <dependency>-->
    42. <!--            <groupId>org.apache.flink</groupId>-->
    43. <!--            <artifactId>flink-table-api-java-bridge_2.11</artifactId>-->
    44. <!--            <version>1.10.0</version>-->
    45. <!--        </dependency>-->
    47. <!--        <dependency>-->
    48. <!--            <groupId>org.apache.flink</groupId>-->
    49. <!--            <artifactId>flink-table</artifactId>-->
    50. <!--            <version>1.10.0</version>-->
    51. <!--        </dependency>-->
    53. <!--        <dependency>-->
    54. <!--            <groupId>org.apache.flink</groupId>-->
    55. <!--            <artifactId>flink-runtime_2.11</artifactId>-->
    56. <!--            <version>1.8.2</version>-->
    57. <!--        </dependency>-->
    59.         <dependency>
    60.             <groupId>org.apache.flink</groupId>
    61.             <artifactId>flink-jdbc_2.11</artifactId>
    62.             <version>1.10.1</version>
    63.         </dependency>
    65. <!--        <dependency>-->
    66. <!--            <groupId>org.apache.flink</groupId>-->
    67. <!--            <artifactId>flink-scala_2.11</artifactId>-->
    68. <!--            <version>1.7.2</version>-->
    69. <!--        </dependency>-->
    71. <!--        <dependency>-->
    72. <!--            <groupId>org.apache.flink</groupId>-->
    73. <!--            <artifactId>flink-java</artifactId>-->
    74. <!--            <version>1.8.2</version>-->
    75. <!--        </dependency>-->
    77. <!--        <dependency>-->
    78. <!--            <groupId>org.apache.flink</groupId>-->
    79. <!--            <artifactId>flink-streaming-scala_2.11</artifactId>-->
    80. <!--            <version>1.7.2</version>-->
    81. <!--        </dependency>-->
    83. <!--        <dependency>-->
    84. <!--            <groupId>org.apache.flink</groupId>-->
    85. <!--            <artifactId>flink-streaming-java_2.11</artifactId>-->
    86. <!--            <version>1.7.2</version>-->
    87. <!--        </dependency>-->
    90.         <dependency>
    91.             <groupId>org.apache.flink</groupId>
    92.             <artifactId>flink-streaming-java_2.11</artifactId>
    93.             <version>1.10.0</version>
    94.         </dependency>
    95.         <dependency>
    96.             <groupId>org.projectlombok</groupId>
    97.             <artifactId>lombok</artifactId>
    98.             <version>1.18.4</version>
    99.         </dependency>
    100.         <dependency>
    101.             <groupId>org.slf4j</groupId>
    102.             <artifactId>slf4j-simple</artifactId>
    103.             <version>1.7.25</version>
    104.         </dependency>
    105.         <dependency>
    106.             <groupId>org.apache.flink</groupId>
    107.             <artifactId>flink-table-api-java</artifactId>
    108.             <version>1.10.0</version>
    109.         </dependency>
    110.         <dependency>
    111.             <groupId>org.apache.flink</groupId>
    112.             <artifactId>flink-table-api-java-bridge_2.11</artifactId>
    113.             <version>1.10.0</version>
    114.         </dependency>
    115.         <dependency>
    116.             <groupId>org.apache.flink</groupId>
    117.             <artifactId>flink-table-planner-blink_2.11</artifactId>
    118.             <version>1.10.0</version>
    119.         </dependency>
    120.         <dependency>
    121.             <groupId>org.apache.flink</groupId>
    122.             <artifactId>flink-table-runtime-blink_2.11</artifactId>
    123.             <version>1.10.0</version>
    124.         </dependency>
    125.         <dependency>
    126.             <groupId>org.apache.flink</groupId>
    127.             <artifactId>flink-table-common</artifactId>
    128.             <version>1.10.0</version>
    129.         </dependency>
    132.     </dependencies>
    137. </project>
    1. import org.apache.flink.api.common.typeinfo.TypeInformation;
    2. import org.apache.flink.api.common.typeinfo.Types;
    3. import org.apache.flink.api.java.io.jdbc.JDBCAppendTableSink;
    4. import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
    5. import org.apache.flink.table.api.EnvironmentSettings;
    6. import org.apache.flink.table.api.Table;
    7. import org.apache.flink.table.api.java.StreamTableEnvironment;
    9. public class FlinkSinkClickHouse {
    10.     public static void main(String[] args) {
    11.         String url = "jdbc:clickhouse://localhost:8123/test";
    12.         String user = "";
    13.         String passwd = "";
    14.         String driver = "ru.yandex.clickhouse.ClickHouseDriver";
    15.         int batchsize = 500; // 设置batch size，测试的话可以设置小一点，这样可以立刻看到数据被写入
    17.         EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
    18.         StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
    19.         StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
    22.         String query = "select 3,'jk'";
    23.         Table table = bsTableEnv.sqlQuery(query);
    25.         String insertIntoCkSql = "INSERT INTO user_local(id,name)\n" +
    26.                 "VALUES(?,?)";
    29.         JDBCAppendTableSink sink = JDBCAppendTableSink
    30.                 .builder()
    31.                 .setDrivername(driver)
    32.                 .setDBUrl(url)
    33.                 .setUsername(user)
    34.                 .setPassword(passwd)
    35.                 .setQuery(insertIntoCkSql)
    36.                 .setBatchSize(batchsize)
    37.                 .setParameterTypes(Types.LONG, Types.STRING)
    38.                 .build();
    40.         String[] arr = {"id","name"};
    41.         TypeInformation[] type = {Types.LONG, Types.STRING};
    43.         bsTableEnv.registerTableSink(
    44.                 "sink",
    45.                 arr,
    46.                 type,
    47.                 sink
    48.         );
    50.         bsTableEnv.insertInto(table, "sink");
    52.         try {
    53.             bsTableEnv.execute("Flink Table API to ClickHouse Example");
    54.         } catch (Exception e) {
    55.             e.printStackTrace();
    56.         }
    60.     }
    62. //    public static void main(String[] args) {
    63. //
    64. //        String url = "jdbc:clickhouse://localhost:8123/test";
    65. //        String user = "";
    66. //        String passwd = "";
    67. //        String driver = "ru.yandex.clickhouse.ClickHouseDriver";
    68. //        int batchsize = 500; // 设置batch size，测试的话可以设置小一点，这样可以立刻看到数据被写入
    69. //
    70. //
    71. //        // 创建执行环境
    72. ////        EnvironmentSettings settings = EnvironmentSettings
    73. ////                .newInstance()
    74. ////                .useBlinkPlanner()
    75. ////                .inStreamingMode()
    76. ////                .build();
    77. //
    78. ////        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    79. //        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    80. //
    81. //        StreamTableEnvironment tbEnv = TableEnvironment(env);
    82. //
    83. //
    84. ////        String kafkaSource11 = "" +
    85. ////                "CREATE TABLE user_behavior ( " +
    86. ////                " `user_id` BIGINT, -- 用户id\n" +
    87. ////                " `item_id` BIGINT, -- 商品id\n" +
    88. ////                " `cat_id` BIGINT, -- 品类id\n" +
    89. ////                " `action` STRING, -- 用户行为\n" +
    90. ////                " `province` INT, -- 用户所在的省份\n" +
    91. ////                " `ts` BIGINT, -- 用户行为发生的时间戳\n" +
    92. ////                " `proctime` AS PROCTIME(), -- 通过计算列产生一个处理时间列\n" +
    93. ////                " `eventTime` AS TO_TIMESTAMP(FROM_UNIXTIME(ts, 'yyyy-MM-dd HH:mm:ss')), -- 事件时间\n" +
    94. ////                " WATERMARK FOR eventTime AS eventTime - INTERVAL '5' SECOND -- 在eventTime上定义watermark\n" +
    95. ////                ") WITH ( 'connector' = 'kafka', -- 使用 kafka connector\n" +
    96. ////                " 'topic' = 'user_behavior', -- kafka主题\n" +
    97. ////                " 'scan.startup.mode' = 'earliest-offset', -- 偏移量，从起始 offset 开始读取\n" +
    98. ////                " 'properties.group.id' = 'group1', -- 消费者组\n" +
    99. ////                " 'properties.bootstrap.servers' = 'kms-2:9092,kms-3:9092,kms-4:9092', -- kafka broker 地址\n" +
    100. ////                " 'format' = 'json', -- 数据源格式为 json\n" +
    101. ////                " 'json.fail-on-missing-field' = 'true',\n" +
    102. ////                " 'json.ignore-parse-errors' = 'false'" +
    103. ////                ")";
    104. ////
    105. ////        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
    106. ////
    107. ////
    108. ////        // Kafka Source
    109. ////        try {
    110. ////            tEnv.execute(kafkaSource11);
    111. ////        } catch (Exception e) {
    112. ////            e.printStackTrace();
    113. ////        }
    114. ////        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
    115. //
    116. //        String query = "select 1,'1';";
    117. //        Table table = tEnv.sqlQuery(query);
    118. //
    119. ////        -- INSERT INTO user_local VALUES(1,'tom'),(2,'jack');
    120. //        String insertIntoCkSql = "INSERT INTO test.user_local(id,name)\n" +
    121. //                "VALUES(?,?)";
    122. //
    123. //        //将数据写入 ClickHouse Sink
    124. //        JDBCAppendTableSink sink = JDBCAppendTableSink
    125. //                .builder()
    126. //                .setDrivername(driver)
    127. //                .setDBUrl(url)
    128. //                .setUsername(user)
    129. //                .setPassword(passwd)
    130. //                .setQuery(insertIntoCkSql)
    131. //                .setBatchSize(batchsize)
    132. //                .setParameterTypes(Types.LONG, Types.STRING)
    133. //                .build();
    134. //
    135. //        String[] arr = {"id","name"};
    136. //        TypeInformation[] type = {Types.LONG, Types.STRING};
    137. //
    138. //        tEnv.registerTableSink(
    139. //                "sink",
    140. //                arr,
    141. //                type,
    142. //                sink
    143. //        );
    144. //
    145. //        tEnv.insertInto(table, "sink");
    146. //
    147. //        try {
    148. //            tEnv.execute("Flink Table API to ClickHouse Example");
    149. //        } catch (Exception e) {
    150. //            e.printStackTrace();
    151. //        }
    152. //    }
    155. }