Skip to content

Commit

Permalink
No51 Constructured Data type
Browse files Browse the repository at this point in the history
  • Loading branch information
@sunjincheng121
sunjincheng121 committed Jun 11, 2021
1 parent 49ef87c commit c5d3e2a
Show file tree
Hide file tree
Showing 10 changed files with 544 additions and 1 deletion.
110 changes: 110 additions & 0 deletions khkw/No46-SQL-Datatypes-Constructured/dependency-reduced-pom.xml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,110 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>khkw</groupId>
<artifactId>No46-SQL-Datatypes-Constructured</artifactId>
<name>Flink Walkthrough DataStream Java</name>
<version>0.1</version>
<url>https://flink.apache.org</url>
<build>
<pluginManagement>
<plugins>
<plugin>
<groupId>org.eclipse.m2e</groupId>
<artifactId>lifecycle-mapping</artifactId>
<version>1.0.0</version>
<configuration>
<lifecycleMappingMetadata>
<pluginExecutions>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<versionRange>[3.0.0,)</versionRange>
<goals>
<goal>shade</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore />
</action>
</pluginExecution>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<versionRange>[3.1,)</versionRange>
<goals>
<goal>testCompile</goal>
<goal>compile</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore />
</action>
</pluginExecution>
</pluginExecutions>
</lifecycleMappingMetadata>
</configuration>
</plugin>
</plugins>
</pluginManagement>
<plugins>
<plugin>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
<plugin>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:force-shading</exclude>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>org.apache.logging.log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer>
<mainClass>datatype.FraudDetectionJob</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<properties>
<scala.binary.version>2.11</scala.binary.version>
<maven.compiler.target>${java.version}</maven.compiler.target>
<java.version>1.8</java.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<log4j.version>2.12.1</log4j.version>
<flink.version>1.13.1</flink.version>
</properties>
</project>

194 changes: 194 additions & 0 deletions khkw/No46-SQL-Datatypes-Constructured/pom.xml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,194 @@
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<groupId>khkw</groupId>
<artifactId>No46-SQL-Datatypes-Constructured</artifactId>
<version>0.1</version>
<packaging>jar</packaging>

<name>Flink Walkthrough DataStream Java</name>
<url>https://flink.apache.org</url>

<properties>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<flink.version>1.13.1</flink.version>
<java.version>1.8</java.version>
<scala.binary.version>2.11</scala.binary.version>
<maven.compiler.source>${java.version}</maven.compiler.source>
<maven.compiler.target>${java.version}</maven.compiler.target>
<log4j.version>2.12.1</log4j.version>
</properties>

<dependencies>

<!-- 利用Java开发 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-api-java</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- 使用Blink Planner -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>

<!-- 支持一些自定义的消息格式 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-table-common</artifactId>
<version>${flink.version}</version>
</dependency>

<!-- 消息采用json & csv格式 -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-json</artifactId>
<version>${flink.version}</version>
</dependency>

<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-csv</artifactId>
<version>${flink.version}</version>
</dependency>

<!--提交作业所必须的依赖，比如：LocalExecutorFactory -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_${scala.binary.version}</artifactId>
<version>${flink.version}</version>
</dependency>

</dependencies>

<build>
<plugins>

<!-- Java Compiler -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.1</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>

<!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
<!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<version>3.0.0</version>
<executions>
<!-- Run shade goal on package phase -->
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<artifactSet>
<excludes>
<exclude>org.apache.flink:force-shading</exclude>
<exclude>com.google.code.findbugs:jsr305</exclude>
<exclude>org.slf4j:*</exclude>
<exclude>org.apache.logging.log4j:*</exclude>
</excludes>
</artifactSet>
<filters>
<filter>
<!-- Do not copy the signatures in the META-INF folder.
Otherwise, this might cause SecurityExceptions when using the JAR. -->
<artifact>*:*</artifact>
<excludes>
<exclude>META-INF/*.SF</exclude>
<exclude>META-INF/*.DSA</exclude>
<exclude>META-INF/*.RSA</exclude>
</excludes>
</filter>
</filters>
<transformers>
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
<mainClass>datatype.FraudDetectionJob</mainClass>
</transformer>
</transformers>
</configuration>
</execution>
</executions>
</plugin>
</plugins>

<pluginManagement>
<plugins>

<!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. -->
<plugin>
<groupId>org.eclipse.m2e</groupId>
<artifactId>lifecycle-mapping</artifactId>
<version>1.0.0</version>
<configuration>
<lifecycleMappingMetadata>
<pluginExecutions>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<versionRange>[3.0.0,)</versionRange>
<goals>
<goal>shade</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore/>
</action>
</pluginExecution>
<pluginExecution>
<pluginExecutionFilter>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<versionRange>[3.1,)</versionRange>
<goals>
<goal>testCompile</goal>
<goal>compile</goal>
</goals>
</pluginExecutionFilter>
<action>
<ignore/>
</action>
</pluginExecution>
</pluginExecutions>
</lifecycleMappingMetadata>
</configuration>
</plugin>
</plugins>
</pluginManagement>
</build>
</project>
65 changes: 65 additions & 0 deletions khkw/No46-SQL-Datatypes-Constructured/src/main/java/datatype/JSONConstructured.java
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
package datatype;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
* 项目名称: Apache Flink 知其然，知其所以然 - datatype.constructured
* 功能描述: 测试 Constructured DataType
* 操作步骤:
* <p>
* 作者： 孙金城
* 日期： 2021/06/11
*/
public class JSONConstructured {
public static void main(String[] args) {
String sourceDDL = "CREATE TABLE json_source (\n" +
" ts BIGINT,\n" +
" name VARCHAR, \n" +
" b_date DATE, \n" +
" tag BOOLEAN, \n" +
" strings ARRAY<VARCHAR>, \n" +
" obj_map MAP<VARCHAR,VARCHAR>, \n" +
" obj_set MULTISET<VARCHAR>, \n" +
" obj_row ROW<f1 BOOLEAN, f2 INT> \n" +
") WITH (\n" +
" 'connector' = 'filesystem' ,\n" +
" 'path' = 'file:///Users/jincheng/work/know_how_know_why/khkw/No46-SQL-Datatypes-Constructured/src/main/resources/mydata.jsonl',\n" +
" 'format' = 'json' ,\n" +
" 'json.timestamp-format.standard' = 'SQL' , \n" + // https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/formats/json/
" 'json.ignore-parse-errors' = 'false' \n" +
")";

String sinkDDL = "CREATE TABLE print_sink (\n" +
" ts BIGINT,\n" +
" name VARCHAR, \n" +
" b_date DATE, \n" +
" tag BOOLEAN, \n" +
" strings ARRAY<VARCHAR>, \n" +
" obj_map MAP<VARCHAR,VARCHAR>, \n" +
" obj_set MULTISET<VARCHAR>, \n" +
" obj_row ROW<f1 BOOLEAN, f2 INT> \n" +
") WITH (\n" +
" 'connector' = 'print'\n" +
")";

// 创建执行环境
EnvironmentSettings settings = EnvironmentSettings.newInstance()
.useBlinkPlanner()
.inStreamingMode()
.build();
StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(sEnv, settings);

//注册source和sink
tEnv.executeSql(sourceDDL);
tEnv.executeSql(sinkDDL);

// SQL逻辑
String sql = "INSERT INTO print_sink SELECT * FROM json_source ";
tEnv.executeSql(sql);

}
}

Loading

0 comments on commit c5d3e2a

Please sign in to comment.