Skip to content

Commit 88bba74

Browse files
viren-nadkarniviren-nadkarni
authored
Add Flink sample that outputs to stdout
2 parents a64439e + 13dbede commit 88bba74

File tree

7 files changed

+351
-2
lines changed

7 files changed

+351
-2
lines changed

‎java/Printer/README.md

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,10 @@
1+
# Printer
2+
3+
* Flink version: 1.20
4+
* Flink API: DataStream API
5+
* Language Java (11)
6+
7+
The Flink application uses a synthetic source to generate ticker records,
8+
parses them, and just prints the results.
9+
10+
It is based on the S3Sink application in this repo.

‎java/Printer/pom.xml

Lines changed: 184 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,184 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
3+
xmlns="http://maven.apache.org/POM/4.0.0"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
7+
<groupId>com.amazonaws</groupId>
8+
<artifactId>msaf-printer</artifactId>
9+
<version>1.0-SNAPSHOT</version>
10+
11+
<distributionManagement>
12+
<repository>
13+
<id>github</id>
14+
<name>GitHub Packages</name>
15+
<url>https://maven.pkg.github.com/localstack-samples/amazon-managed-service-for-apache-flink-examples</url>
16+
</repository>
17+
</distributionManagement>
18+
19+
<properties>
20+
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
21+
<buildDirectory>${project.basedir}/target</buildDirectory>
22+
<jar.finalName>${project.name}</jar.finalName>
23+
<target.java.version>11</target.java.version>
24+
<maven.compiler.source>${target.java.version}</maven.compiler.source>
25+
<maven.compiler.target>${target.java.version}</maven.compiler.target>
26+
<flink.version>1.20.0</flink.version>
27+
<flink.connector.version>5.0.0-1.20</flink.connector.version>
28+
<kda.runtime.version>1.2.0</kda.runtime.version>
29+
<log4j.version>2.23.1</log4j.version>
30+
</properties>
31+
32+
<dependencyManagement>
33+
<dependencies>
34+
<dependency>
35+
<groupId>com.amazonaws</groupId>
36+
<artifactId>aws-java-sdk-bom</artifactId>
37+
<!-- Get the latest SDK version from https://mvnrepository.com/artifact/com.amazonaws/aws-java-sdk-bom -->
38+
<version>1.12.677</version>
39+
<type>pom</type>
40+
<scope>import</scope>
41+
</dependency>
42+
</dependencies>
43+
</dependencyManagement>
44+
45+
<dependencies>
46+
<!-- Apache Flink dependencies -->
47+
<!-- These dependencies are provided, because they should not be packaged into the JAR file. -->
48+
<dependency>
49+
<groupId>org.apache.flink</groupId>
50+
<artifactId>flink-streaming-java</artifactId>
51+
<version>${flink.version}</version>
52+
<scope>provided</scope>
53+
</dependency>
54+
<dependency>
55+
<groupId>org.apache.flink</groupId>
56+
<artifactId>flink-clients</artifactId>
57+
<version>${flink.version}</version>
58+
<scope>provided</scope>
59+
</dependency>
60+
<dependency>
61+
<groupId>org.apache.flink</groupId>
62+
<artifactId>flink-runtime-web</artifactId>
63+
<version>${flink.version}</version>
64+
<scope>provided</scope>
65+
</dependency>
66+
67+
<!-- Library to retrieve runtime application properties in Managed Service for Apache Flink -->
68+
<dependency>
69+
<groupId>com.amazonaws</groupId>
70+
<artifactId>aws-kinesisanalytics-runtime</artifactId>
71+
<version>${kda.runtime.version}</version>
72+
<scope>provided</scope>
73+
</dependency>
74+
75+
<!-- Flink connectors -->
76+
<dependency>
77+
<groupId>org.apache.flink</groupId>
78+
<artifactId>flink-connector-datagen</artifactId>
79+
<version>${flink.version}</version>
80+
</dependency>
81+
<dependency>
82+
<groupId>org.apache.flink</groupId>
83+
<artifactId>flink-connector-files</artifactId>
84+
<version>${flink.version}</version>
85+
<scope>provided</scope>
86+
</dependency>
87+
<dependency>
88+
<groupId>org.apache.flink</groupId>
89+
<artifactId>flink-s3-fs-hadoop</artifactId>
90+
<version>${flink.version}</version>
91+
</dependency>
92+
<dependency>
93+
<groupId>org.apache.flink</groupId>
94+
<artifactId>flink-connector-base</artifactId>
95+
<version>${flink.version}</version>
96+
<scope>provided</scope>
97+
</dependency>
98+
99+
<!-- Add logging framework, to produce console output when running in the IDE. -->
100+
<!-- These dependencies are excluded from the application JAR by default. -->
101+
<dependency>
102+
<groupId>org.apache.logging.log4j</groupId>
103+
<artifactId>log4j-slf4j-impl</artifactId>
104+
<version>${log4j.version}</version>
105+
</dependency>
106+
<dependency>
107+
<groupId>org.apache.logging.log4j</groupId>
108+
<artifactId>log4j-api</artifactId>
109+
<version>${log4j.version}</version>
110+
</dependency>
111+
<dependency>
112+
<groupId>org.apache.logging.log4j</groupId>
113+
<artifactId>log4j-core</artifactId>
114+
<version>${log4j.version}</version>
115+
</dependency>
116+
</dependencies>
117+
118+
<build>
119+
<directory>${buildDirectory}</directory>
120+
<finalName>${jar.finalName}</finalName>
121+
122+
<plugins>
123+
<!-- Java Compiler -->
124+
<plugin>
125+
<groupId>org.apache.maven.plugins</groupId>
126+
<artifactId>maven-compiler-plugin</artifactId>
127+
<version>3.8.1</version>
128+
<configuration>
129+
<source>${target.java.version}</source>
130+
<target>${target.java.version}</target>
131+
<forceJavacCompilerUse>true</forceJavacCompilerUse>
132+
</configuration>
133+
</plugin>
134+
<!-- We use the maven-shade plugin to create a fat jar that contains all necessary dependencies. -->
135+
<!-- Change the value of <mainClass>...</mainClass> if your program entry point changes. -->
136+
<plugin>
137+
<groupId>org.apache.maven.plugins</groupId>
138+
<artifactId>maven-shade-plugin</artifactId>
139+
<version>3.2.1</version>
140+
<executions>
141+
<!-- Run shade goal on package phase -->
142+
<execution>
143+
<phase>package</phase>
144+
<goals>
145+
<goal>shade</goal>
146+
</goals>
147+
<configuration>
148+
<artifactSet>
149+
<excludes>
150+
<exclude>org.apache.flink:force-shading</exclude>
151+
<exclude>com.google.code.findbugs:jsr305</exclude>
152+
<exclude>org.slf4j:*</exclude>
153+
<exclude>org.apache.logging.log4j:*</exclude>
154+
</excludes>
155+
</artifactSet>
156+
<filters>
157+
<filter>
158+
<!-- Do not copy the signatures in the META-INF folder.
159+
Otherwise, this might cause SecurityExceptions when using the JAR. -->
160+
<artifact>*:*</artifact>
161+
<excludes>
162+
<exclude>META-INF/*.SF</exclude>
163+
<exclude>META-INF/*.DSA</exclude>
164+
<exclude>META-INF/*.RSA</exclude>
165+
</excludes>
166+
</filter>
167+
</filters>
168+
<transformers>
169+
<transformer
170+
implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
171+
<transformer
172+
implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
173+
<mainClass>com.amazonaws.services.msf.StreamingJob</mainClass>
174+
</transformer>
175+
</transformers>
176+
</configuration>
177+
</execution>
178+
</executions>
179+
</plugin>
180+
181+
</plugins>
182+
</build>
183+
184+
</project>

‎java/Printer/src/main/java/com/amazonaws/services/msf/StockPrice.java

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
package com.amazonaws.services.msf;
2+
3+
import java.sql.Timestamp;
4+
5+
public class StockPrice {
6+
private Timestamp eventTime;
7+
private String ticker;
8+
private Double price;
9+
10+
11+
public StockPrice() {
12+
}
13+
14+
public StockPrice(Timestamp eventTime, String ticker, Double price) {
15+
this.eventTime = eventTime;
16+
this.ticker = ticker;
17+
this.price = price;
18+
}
19+
20+
public Timestamp getEventTime() {
21+
return eventTime;
22+
}
23+
24+
public void setEventTime(Timestamp eventTime) {
25+
this.eventTime = eventTime;
26+
}
27+
28+
public String getTicker() {
29+
return ticker;
30+
}
31+
32+
public void setTicker(String ticker) {
33+
this.ticker = ticker;
34+
}
35+
36+
public Double getPrice() {
37+
return price;
38+
}
39+
40+
public void setPrice(Double price) {
41+
this.price = price;
42+
}
43+
44+
@Override
45+
public String toString() {
46+
return "StockPrice{" +
47+
"eventTime=" + eventTime +
48+
", ticker='" + ticker + '\'' +
49+
", price=" + price +
50+
'}';
51+
}
52+
}

‎java/Printer/src/main/java/com/amazonaws/services/msf/StockPriceGeneratorFunction.java

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,20 @@
1+
package com.amazonaws.services.msf;
2+
3+
import org.apache.commons.lang3.RandomUtils;
4+
import org.apache.flink.connector.datagen.source.GeneratorFunction;
5+
6+
import java.sql.Timestamp;
7+
import java.time.Instant;
8+
9+
public class StockPriceGeneratorFunction implements GeneratorFunction<Long, StockPrice> {
10+
private static final String[] TICKERS = {"AAPL", "AMZN", "MSFT", "INTC", "TBV"};
11+
12+
@Override
13+
public StockPrice map(Long aLong) {
14+
return new StockPrice(
15+
new Timestamp(Instant.now().toEpochMilli()),
16+
TICKERS[RandomUtils.nextInt(0, TICKERS.length)],
17+
RandomUtils.nextDouble(10,100)
18+
);
19+
}
20+
}

‎java/Printer/src/main/java/com/amazonaws/services/msf/StreamingJob.java

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
package com.amazonaws.services.msf;
2+
3+
import com.fasterxml.jackson.databind.ObjectMapper;
4+
import org.apache.flink.api.common.RuntimeExecutionMode;
5+
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
6+
import org.apache.flink.api.common.typeinfo.TypeInformation;
7+
import org.apache.flink.api.connector.source.util.ratelimit.RateLimiterStrategy;
8+
import org.apache.flink.connector.datagen.source.DataGeneratorSource;
9+
import org.apache.flink.streaming.api.datastream.DataStream;
10+
import org.apache.flink.streaming.api.environment.LocalStreamEnvironment;
11+
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
12+
import org.apache.logging.log4j.LogManager;
13+
import org.apache.logging.log4j.Logger;
14+
15+
import java.io.IOException;
16+
import java.util.Map;
17+
import java.util.Properties;
18+
19+
public class StreamingJob {
20+
21+
private static final Logger LOGGER = LogManager.getLogger(StreamingJob.class);
22+
23+
// Create ObjectMapper instance to serialise POJOs into JSONs
24+
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
25+
26+
private static boolean isLocal(StreamExecutionEnvironment env) {
27+
return env instanceof LocalStreamEnvironment;
28+
}
29+
30+
private static DataGeneratorSource<StockPrice> getStockPriceDataGeneratorSource() {
31+
long recordPerSecond = 100;
32+
return new DataGeneratorSource<>(
33+
new StockPriceGeneratorFunction(),
34+
Long.MAX_VALUE,
35+
RateLimiterStrategy.perSecond(recordPerSecond),
36+
TypeInformation.of(StockPrice.class));
37+
}
38+
39+
public static void main(String[] args) throws Exception {
40+
// Set up the streaming execution environment
41+
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
42+
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
43+
44+
// Local dev specific settings
45+
if (isLocal(env)) {
46+
// Checkpointing and parallelism are set by Amazon Managed Service for Apache Flink when running on AWS
47+
env.enableCheckpointing(60000);
48+
env.setParallelism(2);
49+
}
50+
51+
// Source
52+
DataGeneratorSource<StockPrice> source = getStockPriceDataGeneratorSource();
53+
54+
// DataStream from Source
55+
DataStream<StockPrice> kinesis = env.fromSource(
56+
source, WatermarkStrategy.noWatermarks(), "data-generator").setParallelism(1);
57+
58+
// Print
59+
kinesis.print();
60+
61+
env.execute("Flink Demo Printer Job");
62+
}
63+
}

‎java/Printer/src/main/resources/log4j2.properties

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,12 @@
1+
appender.console.name = ConsoleAppender
2+
appender.console.type = CONSOLE
3+
appender.console.layout.type = PatternLayout
4+
appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %-5p %c{1} - %m%n
5+
6+
#logger.verbose.name = com.amazonaws.services.msf
7+
#logger.verbose.level = debug
8+
#logger.verbose.additivity = false
9+
#logger.verbose.appenderRef.console.ref = ConsoleAppender
10+
11+
rootLogger.level = INFO
12+
rootLogger.appenderRef.console.ref = ConsoleAppender

‎java/S3Sink/pom.xml

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,15 @@
66

77
<groupId>com.amazonaws</groupId>
88
<artifactId>flink-kds-s3</artifactId>
9-
<version>1.0</version>
9+
<version>1.0-SNAPSHOT</version>
10+
11+
<distributionManagement>
12+
<repository>
13+
<id>github</id>
14+
<name>GitHub Packages</name>
15+
<url>https://maven.pkg.github.com/localstack-samples/amazon-managed-service-for-apache-flink-examples</url>
16+
</repository>
17+
</distributionManagement>
1018

1119
<properties>
1220
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
@@ -173,4 +181,4 @@
173181
</plugins>
174182
</build>
175183

176-
</project>
184+
</project>

0 commit comments

Comments
 (0)