high-availability的可选值为NONE或者zookeeper；high-availability.zookeeper.quorum用于指定zookeeper的peers；high-availability.zookeeper.path.root用于指定在zookeeper的root node路径；high-availability.cluster-id用于指定当前cluster的node名称，该cluster node位于root node下面；high-availability.storageD...

使用ParameterTool.fromArgs从命令行创建ParameterTool(比如--input hdfs:///mydata --elements 42)

添加logback-core、logback-classic及log4j-over-slf4j依赖，之后对flink-java、flink-streaming-java_2.11、flink-clients_2.11等配置log4j及slf4j-log4j12的exclusions；最后通过mvn dependency:tree查看是否还有log4j12，以确认下是否都全部排除了

flink-streaming-java_2.11-1.7.1-sources.jar!/org/apache/flink/streaming/api/environment/StreamExecutionEnvironment.java

operators、data sources、data sinks都可以调用setParallelism()方法来设置parallelism

flink-core-1.7.1-sources.jar!/org/apache/flink/api/common/restartstrategy/RestartStrategies.java

flink-table_2.11-1.7.1-sources.jar!/org/apache/flink/table/functions/ScalarFunction.scala

WeightedAvg继承了AggregateFunction，实现了getValue、accumulate、retract、merge、resetAccumulator方法

本实例自定义了Split function，并通过TableEnvironment.registerFunction注册，最后在Table的api或者TableEnvironment.sqlQuery中使用；这里的Split定义了public的eval方法，用于发射数据

本实例定义了MySystemTableSourceFactory，它的requiredContext为update-mode=append及connector.type=my-system，它的supportedProperties为connector.debug，它的createStreamTableSource方法创建的是MySystemAppendTableSource；MySystemConnector继承了ConnectorDescriptor，它定义了connector.type值为my-system，c...

flink-table_2.11-1.7.1-sources.jar!/org/apache/flink/table/sinks/TableSink.scala

flink-table_2.11-1.7.1-sources.jar!/org/apache/flink/table/sources/TableSource.scala

序 本文主要研究一下flink的Table Formats 实例 CSV Format {代码...} flink内置支持csv format，无需添加额外依赖 JSON Format {代码...} 可以使用schema或者jsonSchema或者deriveSchema来定义json format，需要额外添加flink-json依赖 Apache Avro Format {代码...} 可以使用recordClass或者avroSchema来定义Avro sche...

通过timestampsFromField、timestampsFromSource、timestampsFromExtractor定义rowtime

这里使用tableEnv.registerTableSink注册JDBCAppendTableSink，之后利用table.insertInto往该sink写数据

通过TableSource创建Table的话，可以通过实现DefinedProctimeAttribute接口来定义Processing time

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala

序 本文主要研究一下flink Table的Set Operations 实例 Union {代码...} union方法类似sql的union UnionAll {代码...} unionAll方法类似sql的union all Intersect {代码...} intersect方法类似sql的intersect IntersectAll {代码...} intersectAll方法类似sql的intersect all Minus {代码...} minus方法类似sql的except ...

time-windowed join需要至少一个等值条件，然后还需要一个与两边时间相关的条件(可以使用<, <=, >=, >)

Distinct Aggregation可以用于内置的及自定义的aggregation function；内置的aggregation function诸如GroupBy Aggregation、GroupBy Window Aggregation、Over Window Aggregation

Over Windows类似SQL的over子句，它可以基于event-time、processing-time或者row-count；具体可以通过Over类来构造，其中必须设置orderBy、preceding及as方法；它有Unbounded及Bounded两大类

window操作可以对Window进行别名，然后可以在groupBy及select中引用，window有start、end、rowtime属性可以用，其中start及rowtime是inclusive的，而end为exclusive

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/table.scala

flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/api/TableEnvironment.scala

序 本文主要研究一下flink的Table API及SQL Programs 实例 {代码...} 本实例展示了flink的Table API及SQL Programs的基本用法 Table API实例 {代码...} 通过tableEnv.scan方法来创建Table，之后使用Table的各种查询api sqlQuery实例 {代码...} sqlQuery内部是使用Apache Calcite来实现的 sqlUpdate实例(TableSink) {代码...

flink-streaming-java_2.11-1.7.0-sources.jar!/org/apache/flink/streaming/api/operators/async/AsyncWaitOperator.java

本实例展示了flink Async I/O的基本用法，首先是实现AsyncFunction接口，用于编写异步请求逻辑及将结果或异常设置到resultFuture，然后就是使用AsyncDataStream的unorderedWait或orderedWait方法将AsyncFunction作用到DataStream作为transformation；AsyncDataStream的unorderedWait或orderedWait有两个关于async operat...

flink-streaming-java_2.11-1.7.0-sources.jar!/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java