Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit fa7614a

Browse files
feat: yaml parser to read config file
0 parents commit fa7614a

File tree

6 files changed

+428
-0
lines changed

6 files changed

+428
-0
lines changed

‎.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,2 @@
1+
.idea/
2+
target/

‎pom.xml‎

Lines changed: 264 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,264 @@
1+
<!--
2+
Licensed to the Apache Software Foundation (ASF) under one
3+
or more contributor license agreements. See the NOTICE file
4+
distributed with this work for additional information
5+
regarding copyright ownership. The ASF licenses this file
6+
to you under the Apache License, Version 2.0 (the
7+
"License"); you may not use this file except in compliance
8+
with the License. You may obtain a copy of the License at
9+
10+
http://www.apache.org/licenses/LICENSE-2.0
11+
12+
Unless required by applicable law or agreed to in writing,
13+
software distributed under the License is distributed on an
14+
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
KIND, either express or implied. See the License for the
16+
specific language governing permissions and limitations
17+
under the License.
18+
-->
19+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
20+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
21+
<modelVersion>4.0.0</modelVersion>
22+
23+
<groupId>entry</groupId>
24+
<artifactId>kafka-data-pipeline-structured-flink-java</artifactId>
25+
<version>1.0-SNAPSHOT</version>
26+
<packaging>jar</packaging>
27+
28+
<name>Flink Quickstart Job</name>
29+
30+
<properties>
31+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
32+
<flink.version>1.18.1</flink.version>
33+
<target.java.version>1.8</target.java.version>
34+
<scala.binary.version>2.12</scala.binary.version>
35+
<maven.compiler.source>${target.java.version}</maven.compiler.source>
36+
<maven.compiler.target>${target.java.version}</maven.compiler.target>
37+
<log4j.version>2.17.1</log4j.version>
38+
</properties>
39+
40+
<repositories>
41+
<repository>
42+
<id>apache.snapshots</id>
43+
<name>Apache Development Snapshot Repository</name>
44+
<url>https://repository.apache.org/content/repositories/snapshots/</url>
45+
<releases>
46+
<enabled>false</enabled>
47+
</releases>
48+
<snapshots>
49+
<enabled>true</enabled>
50+
</snapshots>
51+
</repository>
52+
</repositories>
53+
54+
<dependencies>
55+
<!-- Apache Flink dependencies -->
56+
<!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
57+
<dependency>
58+
<groupId>org.apache.flink</groupId>
59+
<artifactId>flink-streaming-java</artifactId>
60+
<version>${flink.version}</version>
61+
</dependency>
62+
<dependency>
63+
<groupId>org.apache.flink</groupId>
64+
<artifactId>flink-clients</artifactId>
65+
<version>${flink.version}</version>
66+
</dependency>
67+
<dependency>
68+
<groupId>org.apache.flink</groupId>
69+
<artifactId>flink-connector-kafka</artifactId>
70+
<version>3.0.2-1.18</version>
71+
</dependency>
72+
73+
<!-- Add connector dependencies here. They must be in the default scope (compile). -->
74+
<!-- Example:
75+
76+
<dependency>
77+
<groupId>org.apache.flink</groupId>
78+
<artifactId>flink-connector-kafka</artifactId>
79+
<version>3.0.0-1.17</version>
80+
</dependency>
81+
-->
82+
<dependency>
83+
<groupId>org.apache.flink</groupId>
84+
<artifactId>flink-connector-jdbc</artifactId>
85+
<version>3.0.0-1.16</version>
86+
</dependency>
87+
<dependency>
88+
<groupId>mysql</groupId>
89+
<artifactId>mysql-connector-java</artifactId>
90+
<version>8.0.28</version>
91+
</dependency>
92+
<dependency>
93+
<groupId>com.alibaba.fastjson2</groupId>
94+
<artifactId>fastjson2</artifactId>
95+
<version>2.0.33</version>
96+
</dependency>
97+
<dependency>
98+
<groupId>org.apache.doris</groupId>
99+
<artifactId>flink-doris-connector-1.18</artifactId>
100+
<version>1.6.0</version>
101+
</dependency>
102+
<dependency>
103+
<groupId>org.apache.flink</groupId>
104+
<artifactId>flink-table-api-java</artifactId>
105+
<version>${flink.version}</version>
106+
</dependency>
107+
<dependency>
108+
<groupId>org.apache.flink</groupId>
109+
<artifactId>flink-connector-mongodb</artifactId>
110+
<version>1.1.0-1.18</version>
111+
</dependency>
112+
113+
<!-- Add logging framework, to produce console output when running in the IDE. -->
114+
<!-- These dependencies are excluded from the application JAR by default. -->
115+
<dependency>
116+
<groupId>org.apache.logging.log4j</groupId>
117+
<artifactId>log4j-slf4j-impl</artifactId>
118+
<version>${log4j.version}</version>
119+
<scope>runtime</scope>
120+
</dependency>
121+
<dependency>
122+
<groupId>org.apache.logging.log4j</groupId>
123+
<artifactId>log4j-api</artifactId>
124+
<version>${log4j.version}</version>
125+
<scope>runtime</scope>
126+
</dependency>
127+
<dependency>
128+
<groupId>org.apache.logging.log4j</groupId>
129+
<artifactId>log4j-core</artifactId>
130+
<version>${log4j.version}</version>
131+
<scope>runtime</scope>
132+
</dependency>
133+
<dependency>
134+
<groupId>org.apache.flink</groupId>
135+
<artifactId>flink-connector-base</artifactId>
136+
<version>1.17.1</version>
137+
</dependency>
138+
<dependency>
139+
<groupId>org.projectlombok</groupId>
140+
<artifactId>lombok</artifactId>
141+
<version>1.18.28</version>
142+
</dependency>
143+
</dependencies>
144+
145+
<build>
146+
<finalName>flinkKafka</finalName>
147+
<plugins>
148+
<plugin>
149+
<groupId>org.apache.maven.plugins</groupId>
150+
<artifactId>maven-jar-plugin</artifactId>
151+
<version>3.3.0</version>
152+
<configuration>
153+
<archive>
154+
<manifest>
155+
<addClasspath>true</addClasspath>
156+
<mainClass>your.main.class.MainClass</mainClass>
157+
</manifest>
158+
</archive>
159+
<outputDirectory>${project.build.directory}</outputDirectory>
160+
</configuration>
161+
</plugin>
162+
163+
<!-- Java Compiler -->
164+
<plugin>
165+
<groupId>org.apache.maven.plugins</groupId>
166+
<artifactId>maven-compiler-plugin</artifactId>
167+
<version>3.1</version>
168+
<configuration>
169+
<source>9</source>
170+
<target>9</target>
171+
</configuration>
172+
</plugin>
173+
174+
<!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
175+
<!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
176+
<plugin>
177+
<groupId>org.apache.maven.plugins</groupId>
178+
<artifactId>maven-shade-plugin</artifactId>
179+
<version>3.1.1</version>
180+
<executions>
181+
<!-- Run shade goal on package phase -->
182+
<execution>
183+
<phase>package</phase>
184+
<goals>
185+
<goal>shade</goal>
186+
</goals>
187+
<configuration>
188+
<createDependencyReducedPom>false</createDependencyReducedPom>
189+
<artifactSet>
190+
<excludes>
191+
<exclude>org.apache.flink:flink-shaded-force-shading</exclude>
192+
<exclude>com.google.code.findbugs:jsr305</exclude>
193+
</excludes>
194+
</artifactSet>
195+
<filters>
196+
<filter>
197+
<!-- Do not copy the signatures in the META-INF folder.
198+
Otherwise, this might cause SecurityExceptions when using the JAR. -->
199+
<artifact>*:*</artifact>
200+
<excludes>
201+
<exclude>META-INF/*.SF</exclude>
202+
<exclude>META-INF/*.DSA</exclude>
203+
<exclude>META-INF/*.RSA</exclude>
204+
</excludes>
205+
</filter>
206+
</filters>
207+
<transformers>
208+
<transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
209+
<transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
210+
<mainClass>entry.KafkaToKafka</mainClass>
211+
</transformer>
212+
</transformers>
213+
</configuration>
214+
</execution>
215+
</executions>
216+
</plugin>
217+
</plugins>
218+
219+
<pluginManagement>
220+
<plugins>
221+
222+
<!-- This improves the out-of-the-box experience in Eclipse by resolving some warnings. -->
223+
<plugin>
224+
<groupId>org.eclipse.m2e</groupId>
225+
<artifactId>lifecycle-mapping</artifactId>
226+
<version>1.0.0</version>
227+
<configuration>
228+
<lifecycleMappingMetadata>
229+
<pluginExecutions>
230+
<pluginExecution>
231+
<pluginExecutionFilter>
232+
<groupId>org.apache.maven.plugins</groupId>
233+
<artifactId>maven-shade-plugin</artifactId>
234+
<versionRange>[3.1.1,)</versionRange>
235+
<goals>
236+
<goal>shade</goal>
237+
</goals>
238+
</pluginExecutionFilter>
239+
<action>
240+
<ignore/>
241+
</action>
242+
</pluginExecution>
243+
<pluginExecution>
244+
<pluginExecutionFilter>
245+
<groupId>org.apache.maven.plugins</groupId>
246+
<artifactId>maven-compiler-plugin</artifactId>
247+
<versionRange>[3.1,)</versionRange>
248+
<goals>
249+
<goal>testCompile</goal>
250+
<goal>compile</goal>
251+
</goals>
252+
</pluginExecutionFilter>
253+
<action>
254+
<ignore/>
255+
</action>
256+
</pluginExecution>
257+
</pluginExecutions>
258+
</lifecycleMappingMetadata>
259+
</configuration>
260+
</plugin>
261+
</plugins>
262+
</pluginManagement>
263+
</build>
264+
</project>
Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
package com.examples.job;
2+
3+
import com.examples.parser.CmdLineArgsParser;
4+
import com.examples.parser.YamlParser;
5+
import org.apache.flink.api.common.RuntimeExecutionMode;
6+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
7+
import java.io.*;
8+
import java.net.URL;
9+
10+
public abstract class AbstractFlinkPipelineJob {
11+
protected final StreamExecutionEnvironment env;
12+
protected String jobName;
13+
protected final CmdLineArgsParser cmdLineArgs;
14+
15+
public AbstractFlinkPipelineJob(String[] args) {
16+
this.env = StreamExecutionEnvironment.getExecutionEnvironment();
17+
this.env.enableCheckpointing(10000);
18+
this.env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
19+
this.cmdLineArgs = new CmdLineArgsParser(args);
20+
}
21+
22+
public YamlParser loadSetting() throws IOException {
23+
if (cmdLineArgs.getFilePath() != null) {
24+
String path = cmdLineArgs.getFilePath();
25+
InputStream inputStream = null;
26+
27+
try {
28+
if (path.startsWith("http://") || path.startsWith("https://")) {
29+
URL url = new URL(path);
30+
inputStream = url.openStream();
31+
} else {
32+
File file = new File(path);
33+
inputStream = new FileInputStream(file);
34+
}
35+
36+
YamlParser yamlSetting = new YamlParser(inputStream);
37+
this.jobName = yamlSetting.jobName;
38+
return yamlSetting;
39+
} catch (Exception e) {
40+
throw new RuntimeException(e);
41+
} finally {
42+
if (inputStream != null) {
43+
try {
44+
inputStream.close();
45+
} catch (IOException e) {
46+
e.printStackTrace();
47+
}
48+
}
49+
}
50+
51+
}
52+
return null;
53+
}
54+
55+
protected void run() throws Exception {
56+
this.env.execute(this.jobName);
57+
}
58+
}
Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
package com.examples.parser;
2+
3+
import lombok.Getter;
4+
import org.apache.commons.cli.*;
5+
6+
@Getter
7+
public class CmdLineArgsParser {
8+
private final Options options;
9+
private String filePath;
10+
11+
public CmdLineArgsParser(String[] args) {
12+
this.options = new Options();
13+
options.addOption("f", "file", true, "config file path");
14+
15+
CommandLineParser parser = new DefaultParser();
16+
HelpFormatter formatter = new HelpFormatter();
17+
CommandLine cmd = null;
18+
19+
try {
20+
// Parse the command line arguments
21+
cmd = parser.parse(options, args);
22+
} catch (ParseException e) {
23+
e.printStackTrace();
24+
System.exit(1);
25+
}
26+
27+
if (cmd.hasOption("f")) {
28+
this.filePath = cmd.getOptionValue("f");
29+
} else {
30+
formatter.printHelp("should provide config file path", options);
31+
System.exit(1);
32+
}
33+
}
34+
}

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /