After found matching a b c1 c2 c3 d, the match process will not discard any result. |
|||
SKIP_TO_FIRST[b*] |
+ SKIP_TO_FIRST[c*] |
a b c1 c2 c3 dc1 c2 c3 d@@ -1383,7 +1384,7 @@ Pattern: `a b+` and sequence: `a b1 b2 b3` Then the results will be: | After found matching a b1, the match process will not discard any result. |
SKIP_TO_NEXT[b*] |
+ SKIP_TO_NEXT |
a b1 |
@@ -1509,6 +1510,7 @@ class MyPatternProcessFunction|
| + {% highlight java %} +NUMERIC.quarter +NUMERIC.quarters +{% endhighlight %} + | +
+ Creates an interval of months for NUMERIC quarters. +E.g., |
+ ||
| {% highlight java %} @@ -3586,6 +3597,19 @@ NUMERIC.months | |||
| + {% highlight java %} +NUMERIC.week +NUMERIC.weeks +{% endhighlight %} + | +
+ Creates an interval of milliseconds for NUMERIC weeks. +E.g., |
+ ||
| {% highlight java %} @@ -3831,6 +3855,19 @@ NUMERIC.years | |||
| + {% highlight scala %} +NUMERIC.quarter +NUMERIC.quarters +{% endhighlight %} + | +
+ Creates an interval of months for NUMERIC quarters. +E.g., |
+ ||
| {% highlight scala %} @@ -3843,6 +3880,19 @@ NUMERIC.months | |||
| + {% highlight scala %} +NUMERIC.week +NUMERIC.weeks +{% endhighlight %} + | +
+ Creates an interval of milliseconds for NUMERIC weeks. +E.g., |
+ ||
|
{% highlight scala %}
diff --git a/docs/dev/table/sourceSinks.md b/docs/dev/table/sourceSinks.md
index ab58a725b2ae48..483ef20e64ef4d 100644
--- a/docs/dev/table/sourceSinks.md
+++ b/docs/dev/table/sourceSinks.md
@@ -205,7 +205,7 @@ DefinedRowtimeAttributes {
* `getRowtimeAttributeDescriptors()`: Returns a list of `RowtimeAttributeDescriptor`. A `RowtimeAttributeDescriptor` describes a rowtime attribute with the following properties:
* `attributeName`: The name of the rowtime attribute in the table schema. The field must be defined with type `Types.SQL_TIMESTAMP`.
- * `timestampExtractor`: The timestamp extractor extracts the timestamp from a record with the return type. For example, it can convert convert a Long field into a timestamp or parse a String-encoded timestamp. Flink comes with a set of built-in `TimestampExtractor` implementation for common use cases. It is also possible to provide a custom implementation.
+ * `timestampExtractor`: The timestamp extractor extracts the timestamp from a record with the return type. For example, it can convert a Long field into a timestamp or parse a String-encoded timestamp. Flink comes with a set of built-in `TimestampExtractor` implementation for common use cases. It is also possible to provide a custom implementation.
* `watermarkStrategy`: The watermark strategy defines how watermarks are generated for the rowtime attribute. Flink comes with a set of built-in `WatermarkStrategy` implementations for common use cases. It is also possible to provide a custom implementation.
Attention Although the `getRowtimeAttributeDescriptors()` method returns a list of descriptors, only a single rowtime attribute is support at the moment. We plan to remove this restriction in the future and support tables with more than one rowtime attribute.
diff --git a/docs/dev/table/sql.md b/docs/dev/table/sql.md
index 1e7ef83fcb7237..14df44131f9621 100644
--- a/docs/dev/table/sql.md
+++ b/docs/dev/table/sql.md
@@ -93,7 +93,7 @@ val result2 = tableEnv.sqlQuery(
// SQL update with a registered table
// create and register a TableSink
-TableSink csvSink = new CsvTableSink("/path/to/file", ...)
+val csvSink: CsvTableSink = new CsvTableSink("/path/to/file", ...)
val fieldNames: Array[String] = Array("product", "amount")
val fieldTypes: Array[TypeInformation[_]] = Array(Types.STRING, Types.INT)
tableEnv.registerTableSink("RubberOrders", fieldNames, fieldTypes, csvSink)
diff --git a/docs/dev/table/sqlClient.md b/docs/dev/table/sqlClient.md
index d4eb228c7bf192..97373f409f2fae 100644
--- a/docs/dev/table/sqlClient.md
+++ b/docs/dev/table/sqlClient.md
@@ -272,7 +272,7 @@ execution:
### Dependencies
-The SQL Client does not require to setup a Java project using Maven or SBT. Instead, you can pass the dependencies as regular JAR files that get submitted to the cluster. You can either specify each JAR file separately (using `--jar`) or define entire library directories (using `--library`). For connectors to external systems (such as Apache Kafka) and corresponding data formats (such as JSON), Flink provides **ready-to-use JAR bundles**. These JAR files are suffixed with `sql-jar` and can be downloaded for each release from the Maven central repository.
+The SQL Client does not require to setup a Java project using Maven or SBT. Instead, you can pass the dependencies as regular JAR files that get submitted to the cluster. You can either specify each JAR file separately (using `--jar`) or define entire library directories (using `--library`). For connectors to external systems (such as Apache Kafka) and corresponding data formats (such as JSON), Flink provides **ready-to-use JAR bundles**. These JAR files can be downloaded for each release from the Maven central repository.
The full list of offered SQL JARs and documentation about how to use them can be found on the [connection to external systems page](connect.html).
diff --git a/docs/dev/table/streaming/index.md b/docs/dev/table/streaming/index.md
index 95c6bab8e0a50c..a6fa525a3dbbc8 100644
--- a/docs/dev/table/streaming/index.md
+++ b/docs/dev/table/streaming/index.md
@@ -40,3 +40,5 @@ Where to go next?
* [Joins in Continuous Queries]({{ site.baseurl }}/dev/table/streaming/joins.html): Different supported types of Joins in Continuous Queries.
* [Temporal Tables]({{ site.baseurl }}/dev/table/streaming/temporal_tables.html): Describes the Temporal Table concept.
* [Query configuration]({{ site.baseurl }}/dev/table/streaming/query_configuration.html): Lists Table API & SQL specific configuration options.
+
+{% top %}
diff --git a/docs/dev/table/streaming/joins.md b/docs/dev/table/streaming/joins.md
index 508e8c70221c3e..a572c1be293cf4 100644
--- a/docs/dev/table/streaming/joins.md
+++ b/docs/dev/table/streaming/joins.md
@@ -221,3 +221,5 @@ applied updates according to the primary key until this point in time.
By definition of event time, [watermarks]({{ site.baseurl }}/dev/event_time.html) allow the join operation to move
forward in time and discard versions of the build table that are no longer necessary because no incoming row with
lower or equal timestamp is expected.
+
+{% top %}
diff --git a/docs/dev/table/streaming/match_recognize.md b/docs/dev/table/streaming/match_recognize.md
index 33afc27e75af7b..4090440f8645c0 100644
--- a/docs/dev/table/streaming/match_recognize.md
+++ b/docs/dev/table/streaming/match_recognize.md
@@ -960,3 +960,5 @@ Unsupported features include:
* `MATCH_RECOGNIZE` is supported only for SQL. There is no equivalent in the Table API.
* Aggregations:
* distinct aggregations are not supported.
+
+{% top %}
diff --git a/docs/dev/table/streaming/temporal_tables.md b/docs/dev/table/streaming/temporal_tables.md
index 4ebb4a6e8a775f..d511c1f93f4594 100644
--- a/docs/dev/table/streaming/temporal_tables.md
+++ b/docs/dev/table/streaming/temporal_tables.md
@@ -186,3 +186,5 @@ which allows us to use the function `rates` in the [Table API](../tableApi.html#
Line `(2)` registers this function under the name `Rates` in our table environment,
which allows us to use the `Rates` function in [SQL](../sql.html#joins).
+
+{% top %}
diff --git a/docs/dev/table/streaming/time_attributes.md b/docs/dev/table/streaming/time_attributes.md
index 101bad68b80822..9014d2a0007f14 100644
--- a/docs/dev/table/streaming/time_attributes.md
+++ b/docs/dev/table/streaming/time_attributes.md
@@ -256,15 +256,18 @@ val windowedTable = table.window(Tumble over 10.minutes on 'UserActionTime as 'u
### Using a TableSource
-The event time attribute is defined by a `TableSource` that implements the `DefinedRowtimeAttribute` interface. The `getRowtimeAttribute()` method returns the name of an existing field that carries the event time attribute of the table and is of type `LONG` or `TIMESTAMP`.
+The event time attribute is defined by a `TableSource` that implements the `DefinedRowtimeAttributes` interface. The `getRowtimeAttributeDescriptors()` method returns a list of `RowtimeAttributeDescriptor` for describing the final name of a time attribute, a timestamp extractor to derive the values of the attribute, and the watermark strategy associated with the attribute.
-Moreover, the `DataStream` returned by the `getDataStream()` method must have watermarks assigned that are aligned with the defined time attribute. Please note that the timestamps of the `DataStream` (the ones which are assigned by a `TimestampAssigner`) are ignored. Only the values of the `TableSource`'s rowtime attribute are relevant.
+Please make sure that the `DataStream` returned by the `getDataStream()` method is aligned with the defined time attribute.
+The timestamps of the `DataStream` (the ones which are assigned by a `TimestampAssigner`) are only considered if a `StreamRecordTimestamp` timestamp extractor is defined.
+Watermarks of a `DataStream` are only preserved if a `PreserveWatermarks` watermark strategy is defined.
+Otherwise, only the values of the `TableSource`'s rowtime attribute are relevant.
{% highlight java %}
// define a table source with a rowtime attribute
-public class UserActionSource implements StreamTableSource
{% highlight scala %}
// define a table source with a rowtime attribute
-class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttribute {
+class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttributes {
override def getReturnType = {
val names = Array[String]("Username" , "Data", "UserActionTime")
@@ -317,9 +326,15 @@ class UserActionSource extends StreamTableSource[Row] with DefinedRowtimeAttribu
stream
}
- override def getRowtimeAttribute = {
+ override def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor] = {
// Mark the "UserActionTime" attribute as event-time attribute.
- "UserActionTime"
+ // We create one attribute descriptor of "UserActionTime".
+ val rowtimeAttrDescr = new RowtimeAttributeDescriptor(
+ "UserActionTime",
+ new ExistingField("UserActionTime"),
+ new AscendingTimestamps)
+ val listRowtimeAttrDescr = Collections.singletonList(rowtimeAttrDescr)
+ listRowtimeAttrDescr
}
}
diff --git a/docs/dev/table/tableApi.md b/docs/dev/table/tableApi.md
index e44df24ad5ec96..d44eebddf53519 100644
--- a/docs/dev/table/tableApi.md
+++ b/docs/dev/table/tableApi.md
@@ -109,7 +109,7 @@ Table orders = tEnv.scan("Orders"); // schema (a, b, c, rowtime)
Table result = orders
.filter("a.isNotNull && b.isNotNull && c.isNotNull")
- .select("a.lowerCase(), b, rowtime")
+ .select("a.lowerCase() as a, b, rowtime")
.window(Tumble.over("1.hour").on("rowtime").as("hourlyWindow"))
.groupBy("hourlyWindow, a")
.select("a, hourlyWindow.end as hour, b.avg as avgBillingAmount");
@@ -128,7 +128,7 @@ val orders: Table = tEnv.scan("Orders") // schema (a, b, c, rowtime)
val result: Table = orders
.filter('a.isNotNull && 'b.isNotNull && 'c.isNotNull)
- .select('a.lowerCase(), 'b, 'rowtime)
+ .select('a.lowerCase() as 'a, 'b, 'rowtime)
.window(Tumble over 1.hour on 'rowtime as 'hourlyWindow)
.groupBy('hourlyWindow, 'a)
.select('a, 'hourlyWindow.end as 'hour, 'b.avg as 'avgBillingAmount)
diff --git a/docs/dev/table/udfs.md b/docs/dev/table/udfs.md
index 44b8d431c8d7d0..d93b5c8c3f7972 100644
--- a/docs/dev/table/udfs.md
+++ b/docs/dev/table/udfs.md
@@ -190,7 +190,7 @@ tableEnv.sqlQuery("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(s
class Split(separator: String) extends TableFunction[(String, Int)] {
def eval(str: String): Unit = {
// use collect(...) to emit a row.
- str.split(separator).foreach(x -> collect((x, x.length))
+ str.split(separator).foreach(x => collect((x, x.length)))
}
}
@@ -210,7 +210,7 @@ tableEnv.registerFunction("split", new Split("#"))
// CROSS JOIN a table function (equivalent to "join" in Table API)
tableEnv.sqlQuery("SELECT a, word, length FROM MyTable, LATERAL TABLE(split(a)) as T(word, length)")
// LEFT JOIN a table function (equivalent to "leftOuterJoin" in Table API)
-tableEnv.sqlQuery("SELECT a, word, length FROM MyTable LEFT JOIN TABLE(split(a)) as T(word, length) ON TRUE")
+tableEnv.sqlQuery("SELECT a, word, length FROM MyTable LEFT JOIN LATERAL TABLE(split(a)) as T(word, length) ON TRUE")
{% endhighlight %}
**IMPORTANT:** Do not implement TableFunction as a Scala object. Scala object is a singleton and will cause concurrency issues.
diff --git a/docs/examples/index.md b/docs/examples/index.md
index a8472026b7c4e0..f7242241167e77 100644
--- a/docs/examples/index.md
+++ b/docs/examples/index.md
@@ -45,4 +45,6 @@ There are also a few blog posts published online that discuss example applicatio
* [Building real-time dashboard applications with Apache Flink, Elasticsearch, and Kibana](https://www.elastic.co/blog/building-real-time-dashboard-applications-with-apache-flink-elasticsearch-and-kibana) is a blog post at elastic.co showing how to build a real-time dashboard solution for streaming data analytics using Apache Flink, Elasticsearch, and Kibana.
-* The [Flink training website](http://training.data-artisans.com/) from data Artisans has a number of examples. Check out the hands-on sections and the exercises.
\ No newline at end of file
+* The [Flink training website](http://training.data-artisans.com/) from data Artisans has a number of examples. Check out the hands-on sections and the exercises.
+
+{% top %}
diff --git a/docs/monitoring/debugging_classloading.md b/docs/monitoring/debugging_classloading.md
index fe2c36d1f14fba..52d15280932682 100644
--- a/docs/monitoring/debugging_classloading.md
+++ b/docs/monitoring/debugging_classloading.md
@@ -85,7 +85,6 @@ classes are loaded dynamically when the jobs are submitted.
In setups where dynamic classloading is involved (sessions), there is a hierarchy of typically two ClassLoaders:
(1) Java's *application classloader*, which has all classes in the classpath, and (2) the dynamic *user code classloader*.
for loading classes from the user-code jar(s). The user-code ClassLoader has the application classloader as its parent.
-cases.
By default, Flink inverts classloading order, meaning it looks into the user code classloader first, and only looks into
the parent (application classloader) if the class is not part of the dynamically loaded user code.
diff --git a/docs/monitoring/metrics.md b/docs/monitoring/metrics.md
index bfa0576c98caf7..ae51f9ee3d4422 100644
--- a/docs/monitoring/metrics.md
+++ b/docs/monitoring/metrics.md
@@ -652,6 +652,7 @@ of your Flink distribution.
Parameters:
- `port` - (optional) the port the Prometheus exporter listens on, defaults to [9249](https://github.com/prometheus/prometheus/wiki/Default-port-allocations). In order to be able to run several instances of the reporter on one host (e.g. when one TaskManager is colocated with the JobManager) it is advisable to use a port range like `9250-9260`.
+- `filterLabelValueCharacters` - (optional) Specifies whether to filter label value characters. If enabled, all characters not matching \[a-zA-Z0-9:_\] will be removed, otherwise no characters will be removed. Before disabling this option please ensure that your label values meet the [Prometheus requirements](https://prometheus.io/docs/concepts/data_model/#metric-names-and-labels).
Example configuration:
diff --git a/docs/ops/cli.md b/docs/ops/cli.md
index 9af5c5b5333095..ec3596fa55d2b5 100644
--- a/docs/ops/cli.md
+++ b/docs/ops/cli.md
@@ -30,11 +30,6 @@ distributed setups. It is located under `
+
+
+Note, that using $ as a prefix for the class name of TableFunction is a workaround of the issue that scala incorrectly generated inner class name.
+
+### SQL
+
+The following example is a wordcount program written in SQL:
+
+{% highlight scala %}
+Scala-Flink> import org.apache.flink.table.functions.TableFunction
+Scala-Flink> val textSource = stenv.fromDataStream(
+ senv.fromElements(
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,"),
+ 'text)
+Scala-Flink> class $Split extends TableFunction[String] {
+ def eval(s: String): Unit = {
+ s.toLowerCase.split("\\W+").foreach(collect)
+ }
+ }
+Scala-Flink> val split = new $Split
+Scala-Flink> textSource.join(split('text) as 'word).
+ groupBy('word).select('word, 'word.count as 'count).
+ toRetractStream[(String, Long)].print
+Scala-Flink> senv.execute("Table Wordcount")
+{% endhighlight %}
+
+
+{% highlight scala %}
+Scala-Flink> import org.apache.flink.table.functions.TableFunction
+Scala-Flink> val textSource = btenv.fromDataSet(
+ benv.fromElements(
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,"),
+ 'text)
+Scala-Flink> class $Split extends TableFunction[String] {
+ def eval(s: String): Unit = {
+ s.toLowerCase.split("\\W+").foreach(collect)
+ }
+ }
+Scala-Flink> val split = new $Split
+Scala-Flink> textSource.join(split('text) as 'word).
+ groupBy('word).select('word, 'word.count as 'count).
+ toDataSet[(String, Long)].print
+{% endhighlight %}
+
+
+
## Adding external dependencies
diff --git a/docs/ops/state/large_state_tuning.md b/docs/ops/state/large_state_tuning.md
index 4bf890d622aa3e..81d360660754cd 100644
--- a/docs/ops/state/large_state_tuning.md
+++ b/docs/ops/state/large_state_tuning.md
@@ -132,7 +132,7 @@ Incremental checkpoints can dramatically reduce the checkpointing time in compar
recovery time. The core idea is that incremental checkpoints only record all changes to the previous completed checkpoint, instead of
producing a full, self-contained backup of the state backend. Like this, incremental checkpoints build upon previous checkpoints. Flink leverages
RocksDB's internal backup mechanism in a way that is self-consolidating over time. As a result, the incremental checkpoint history in Flink
-does not grow indefinitely, and old checkpoints are eventually subsumed and pruned automatically. `
+does not grow indefinitely, and old checkpoints are eventually subsumed and pruned automatically.
While we strongly encourage the use of incremental checkpoints for large state, please note that this is a new feature and currently not enabled
by default. To enable this feature, users can instantiate a `RocksDBStateBackend` with the corresponding boolean flag in the constructor set to `true`, e.g.:
diff --git a/docs/ops/state/state_backends.md b/docs/ops/state/state_backends.md
index e4fe5bda063045..c1678036875f75 100644
--- a/docs/ops/state/state_backends.md
+++ b/docs/ops/state/state_backends.md
@@ -146,13 +146,23 @@ env.setStateBackend(new FsStateBackend("hdfs://namenode:40010/flink/checkpoints"
+{% highlight scala %}
+Scala-Flink> import org.apache.flink.table.functions.TableFunction
+Scala-Flink> val textSource = stenv.fromDataStream(
+ senv.fromElements(
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,"),
+ 'text)
+Scala-Flink> stenv.registerTable("text_source", textSource)
+Scala-Flink> class $Split extends TableFunction[String] {
+ def eval(s: String): Unit = {
+ s.toLowerCase.split("\\W+").foreach(collect)
+ }
+ }
+Scala-Flink> stenv.registerFunction("split", new $Split)
+Scala-Flink> val result = stenv.sqlQuery("""SELECT T.word, count(T.word) AS `count`
+ FROM text_source
+ JOIN LATERAL table(split(text)) AS T(word)
+ ON TRUE
+ GROUP BY T.word""")
+Scala-Flink> result.toRetractStream[(String, Long)].print
+Scala-Flink> senv.execute("SQL Wordcount")
+{% endhighlight %}
+
+
+{% highlight scala %}
+Scala-Flink> import org.apache.flink.table.functions.TableFunction
+Scala-Flink> val textSource = btenv.fromDataSet(
+ benv.fromElements(
+ "To be, or not to be,--that is the question:--",
+ "Whether 'tis nobler in the mind to suffer",
+ "The slings and arrows of outrageous fortune",
+ "Or to take arms against a sea of troubles,"),
+ 'text)
+Scala-Flink> btenv.registerTable("text_source", textSource)
+Scala-Flink> class $Split extends TableFunction[String] {
+ def eval(s: String): Unit = {
+ s.toLowerCase.split("\\W+").foreach(collect)
+ }
+ }
+Scala-Flink> btenv.registerFunction("split", new $Split)
+Scala-Flink> val result = btenv.sqlQuery("""SELECT T.word, count(T.word) AS `count`
+ FROM text_source
+ JOIN LATERAL table(split(text)) AS T(word)
+ ON TRUE
+ GROUP BY T.word""")
+Scala-Flink> result.toDataSet[(String, Long)].print
+{% endhighlight %}
+
+
sendRequest(
- final String targetAddress,
- final int targetPort,
- final M messageHeaders,
- final U messageParameters,
- final R request) throws IOException {
+ final String targetAddress,
+ final int targetPort,
+ final M messageHeaders,
+ final U messageParameters,
+ final R request) throws IOException {
if (failHttpRequest.test(messageHeaders, messageParameters, request)) {
return FutureUtils.completedExceptionally(new IOException("expected"));
} else {
@@ -196,26 +221,6 @@ public void setUp() throws Exception {
}
}
};
- restClusterClient = new RestClusterClient<>(
- config,
- restClient,
- StandaloneClusterId.getInstance(),
- (attempt) -> 0,
- null);
-
- jobGraph = new JobGraph("testjob");
- jobId = jobGraph.getJobID();
- }
-
- @After
- public void tearDown() throws Exception {
- if (restClusterClient != null) {
- restClusterClient.shutdown();
- }
-
- if (executor != null) {
- executor.shutdown();
- }
}
@Test
@@ -230,22 +235,27 @@ public void testJobSubmitCancelStop() throws Exception {
.netRuntime(Long.MAX_VALUE)
.build()));
- try (TestRestServerEndpoint ignored = createRestServerEndpoint(
+ try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(
submitHandler,
terminationHandler,
testJobExecutionResultHandler)) {
+ RestClusterClient restClusterClient = createRestClusterClient(restServerEndpoint.getServerAddress().getPort());
- Assert.assertFalse(submitHandler.jobSubmitted);
- restClusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
- Assert.assertTrue(submitHandler.jobSubmitted);
+ try {
+ Assert.assertFalse(submitHandler.jobSubmitted);
+ restClusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
+ Assert.assertTrue(submitHandler.jobSubmitted);
- Assert.assertFalse(terminationHandler.jobCanceled);
- restClusterClient.cancel(jobId);
- Assert.assertTrue(terminationHandler.jobCanceled);
+ Assert.assertFalse(terminationHandler.jobCanceled);
+ restClusterClient.cancel(jobId);
+ Assert.assertTrue(terminationHandler.jobCanceled);
- Assert.assertFalse(terminationHandler.jobStopped);
- restClusterClient.stop(jobId);
- Assert.assertTrue(terminationHandler.jobStopped);
+ Assert.assertFalse(terminationHandler.jobStopped);
+ restClusterClient.stop(jobId);
+ Assert.assertTrue(terminationHandler.jobStopped);
+ } finally {
+ restClusterClient.shutdown();
+ }
}
}
@@ -257,16 +267,21 @@ public void testDetachedJobSubmission() throws Exception {
final TestJobSubmitHandler testJobSubmitHandler = new TestJobSubmitHandler();
- try (TestRestServerEndpoint ignored = createRestServerEndpoint(
+ try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(
testJobSubmitHandler)) {
+ RestClusterClient restClusterClient = createRestClusterClient(restServerEndpoint.getServerAddress().getPort());
- restClusterClient.setDetached(true);
- final JobSubmissionResult jobSubmissionResult = restClusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
-
- // if the detached mode didn't work, then we would not reach this point because the execution result
- // retrieval would have failed.
- assertThat(jobSubmissionResult, is(not(instanceOf(JobExecutionResult.class))));
- assertThat(jobSubmissionResult.getJobID(), is(jobId));
+ try {
+ restClusterClient.setDetached(true);
+ final JobSubmissionResult jobSubmissionResult = restClusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
+
+ // if the detached mode didn't work, then we would not reach this point because the execution result
+ // retrieval would have failed.
+ assertThat(jobSubmissionResult, is(not(instanceOf(JobExecutionResult.class))));
+ assertThat(jobSubmissionResult.getJobID(), is(jobId));
+ } finally {
+ restClusterClient.shutdown();
+ }
}
}
@@ -365,29 +380,34 @@ public void testSubmitJobAndWaitForExecutionResult() throws Exception {
failHttpRequest = (messageHeaders, messageParameters, requestBody) ->
messageHeaders instanceof JobExecutionResultHeaders && !firstPollFailed.getAndSet(true);
- try (TestRestServerEndpoint ignored = createRestServerEndpoint(
+ try (TestRestServerEndpoint restServerEndpoint = createRestServerEndpoint(
testJobExecutionResultHandler,
new TestJobSubmitHandler())) {
+ RestClusterClient restClusterClient = createRestClusterClient(restServerEndpoint.getServerAddress().getPort());
- JobExecutionResult jobExecutionResult;
+ try {
+ JobExecutionResult jobExecutionResult;
- jobExecutionResult = (JobExecutionResult) restClusterClient.submitJob(
- jobGraph,
- ClassLoader.getSystemClassLoader());
- assertThat(jobExecutionResult.getJobID(), equalTo(jobId));
- assertThat(jobExecutionResult.getNetRuntime(), equalTo(Long.MAX_VALUE));
- assertThat(
- jobExecutionResult.getAllAccumulatorResults(),
- equalTo(Collections.singletonMap("testName", 1.0)));
+ jobExecutionResult = (JobExecutionResult) restClusterClient.submitJob(
+ jobGraph,
+ ClassLoader.getSystemClassLoader());
+ assertThat(jobExecutionResult.getJobID(), equalTo(jobId));
+ assertThat(jobExecutionResult.getNetRuntime(), equalTo(Long.MAX_VALUE));
+ assertThat(
+ jobExecutionResult.getAllAccumulatorResults(),
+ equalTo(Collections.singletonMap("testName", 1.0)));
- try {
- restClusterClient.submitJob(jobGraph, ClassLoader.getSystemClassLoader());
- fail("Expected exception not thrown.");
- } catch (final ProgramInvocationException e) {
- final Optional This example shows how to run program directly within Java, thus it cannot be used to submit a
- * {@link org.apache.storm.generated.StormTopology} via Flink command line clients (ie, bin/flink).
- *
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage:
- * This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * This example shows how to:
- * This example would work the same way for multiple bolt output streams.
- */
-public class SpoutSplitExample {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
- public static void main(final String[] args) throws Exception {
-
- boolean useFile = SpoutSplitExample.parseParameters(args);
-
- // set up the execution environment
- final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
-
- String[] rawOutputs = new String[] { RandomSpout.EVEN_STREAM, RandomSpout.ODD_STREAM };
-
- final DataStream The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * This example shows how to run program directly within Java, thus it cannot be used to submit a {@link StormTopology}
- * via Flink command line clients (ie, bin/flink).
- *
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * This example shows how to run program directly within Java, thus it cannot be used to submit a {@link StormTopology}
- * via Flink command line clients (ie, bin/flink).
- *
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * This example shows how to submit the program via Java, thus it cannot be used to submit a {@link StormTopology} via
- * Flink command line clients (ie, bin/flink).
- *
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * This example shows how to submit the program via Java as well as Flink's command line client (ie, bin/flink).
- *
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage: This example shows how to:
- * The input is a plain text file with lines separated by newline characters.
- *
- * Usage:
- * This example shows how to:
- * Same as {@link BoltCounterByName}, but accesses input attribute by index (instead of name).
- */
-public class BoltCounter implements IRichBolt {
- private static final long serialVersionUID = 399619605462625934L;
-
- public static final String ATTRIBUTE_WORD = "word";
- public static final String ATTRIBUTE_COUNT = "count";
-
- private final HashMap Same as {@link BoltCounter}, but accesses input attribute by name (instead of index).
- */
-public class BoltCounterByName implements IRichBolt {
- private static final long serialVersionUID = 399619605462625934L;
-
- public static final String ATTRIBUTE_WORD = "word";
- public static final String ATTRIBUTE_COUNT = "count";
-
- private final HashMap Same as {@link BoltTokenizerByName}, but accesses input attribute by index (instead of name).
- */
-public final class BoltTokenizer implements IRichBolt {
- private static final long serialVersionUID = -8589620297208175149L;
-
- public static final String ATTRIBUTE_WORD = "word";
- public static final String ATTRIBUTE_COUNT = "count";
-
- public static final int ATTRIBUTE_WORD_INDEX = 0;
- public static final int ATTRIBUTE_COUNT_INDEX = 1;
-
- private OutputCollector collector;
-
- @SuppressWarnings("rawtypes")
- @Override
- public void prepare(final Map stormConf, final TopologyContext context, final OutputCollector collector) {
- this.collector = collector;
- }
-
- @Override
- public void execute(final Tuple input) {
- final String[] tokens = input.getString(0).toLowerCase().split("\\W+");
-
- for (final String token : tokens) {
- if (token.length() > 0) {
- this.collector.emit(new Values(token, 1));
- }
- }
- }
-
- @Override
- public void cleanup() {/* nothing to do */}
-
- @Override
- public void declareOutputFields(final OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields(ATTRIBUTE_WORD, ATTRIBUTE_COUNT));
- }
-
- @Override
- public Map Same as {@link BoltTokenizer}, but accesses input attribute by name (instead of index).
- */
-public final class BoltTokenizerByName implements IRichBolt {
- private static final long serialVersionUID = -8589620297208175149L;
-
- public static final String ATTRIBUTE_WORD = "word";
- public static final String ATTRIBUTE_COUNT = "count";
-
- public static final int ATTRIBUTE_WORD_INDEX = 0;
- public static final int ATTRIBUTE_COUNT_INDEX = 1;
-
- private OutputCollector collector;
-
- @SuppressWarnings("rawtypes")
- @Override
- public void prepare(final Map stormConf, final TopologyContext context, final OutputCollector collector) {
- this.collector = collector;
- }
-
- @Override
- public void execute(final Tuple input) {
- final String[] tokens = input.getStringByField("sentence").toLowerCase().split("\\W+");
-
- for (final String token : tokens) {
- if (token.length() > 0) {
- this.collector.emit(new Values(token, 1));
- }
- }
- }
-
- @Override
- public void cleanup() {/* nothing to do */}
-
- @Override
- public void declareOutputFields(final OutputFieldsDeclarer declarer) {
- declarer.declare(new Fields(ATTRIBUTE_WORD, ATTRIBUTE_COUNT));
- }
-
- @Override
- public Map {@link FlinkClient} mimics both, {@link NimbusClient} and {@link Nimbus}{@code .Client}, at once.
- *
- * @return A reference to itself.
- */
- public FlinkClient getClient() {
- return this;
- }
-
- // The following methods are derived from "backtype.storm.generated.Nimbus.Client"
-
- /**
- * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
- * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
- */
- public void submitTopology(final String name, final String uploadedJarLocation, final FlinkTopology topology)
- throws AlreadyAliveException, InvalidTopologyException {
- this.submitTopologyWithOpts(name, uploadedJarLocation, topology);
- }
-
- /**
- * Parameter {@code uploadedJarLocation} is actually used to point to the local jar, because Flink does not support
- * uploading a jar file before hand. Jar files are always uploaded directly when a program is submitted.
- */
- public void submitTopologyWithOpts(final String name, final String uploadedJarLocation, final FlinkTopology topology)
- throws AlreadyAliveException, InvalidTopologyException {
-
- if (this.getTopologyJobId(name) != null) {
- throw new AlreadyAliveException();
- }
-
- final URI uploadedJarUri;
- final URL uploadedJarUrl;
- try {
- uploadedJarUri = new File(uploadedJarLocation).getAbsoluteFile().toURI();
- uploadedJarUrl = uploadedJarUri.toURL();
- JobWithJars.checkJarFile(uploadedJarUrl);
- } catch (final IOException e) {
- throw new RuntimeException("Problem with jar file " + uploadedJarLocation, e);
- }
-
- try {
- FlinkClient.addStormConfigToTopology(topology, conf);
- } catch (ClassNotFoundException e) {
- LOG.error("Could not register class for Kryo serialization.", e);
- throw new InvalidTopologyException("Could not register class for Kryo serialization.");
- }
-
- final StreamGraph streamGraph = topology.getExecutionEnvironment().getStreamGraph();
- streamGraph.setJobName(name);
-
- final JobGraph jobGraph = streamGraph.getJobGraph();
- jobGraph.addJar(new Path(uploadedJarUri));
-
- final Configuration configuration = jobGraph.getJobConfiguration();
- configuration.setString(JobManagerOptions.ADDRESS, jobManagerHost);
- configuration.setInteger(JobManagerOptions.PORT, jobManagerPort);
-
- final StandaloneClusterClient client;
- try {
- client = new StandaloneClusterClient(configuration);
- } catch (final Exception e) {
- throw new RuntimeException("Could not establish a connection to the job manager", e);
- }
-
- try {
- ClassLoader classLoader = JobWithJars.buildUserCodeClassLoader(
- Collections. |