flink-1.7.2/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/NetworkEnvironmentConfiguration.java

flink-1.7.2/flink-core/src/main/java/org/apache/flink/configuration/TaskManagerOptions.java

flink-release-1.7.2/flink-core/src/main/java/org/apache/flink/core/memory/MemorySegment.java

flink-runtime_2.11-1.7.2-sources.jar!/org/apache/flink/runtime/memory/MemoryManager.java

flink-core-1.7.2-sources.jar!/org/apache/flink/configuration/TaskManagerOptions.java

flink-conf.yaml提供了taskmanager.heap.size来设置taskmanager的memory(heap及offHeap)大小

flink-core-1.7.1-sources.jar!/org/apache/flink/configuration/JobManagerOptions.java

flink-runtime_2.11-1.7.1-sources.jar!/org/apache/flink/runtime/highavailability/HighAvailabilityServices.java

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