hello me you her

hello me you

hello me

hello

/export/server/flink/bin/flink run /export/server/flink/examples/batch/WordCount.jar --input /root/words.txt --output /root/out

benv.readTextFile("/root/words.txt").flatMap(_.split(" ")).map((_,1)).groupBy(0).sum(1).print()

jobmanager.rpc.address: node1

taskmanager.numberOfTaskSlots: 2

web.submit.enable: true

#歷史伺服器

jobmanager.archive.fs.dir: hdfs://node1:8020/flink/completed-jobs/

historyserver.web.address: node1

historyserver.web.port: 8082

historyserver.archive.fs.dir: hdfs://node1:8020/flink/completed-jobs/

node1:8081

node1

node2

node3

export HADOOP_CONF_DIR=/export/server/hadoop/etc/hadoop

/export/server/flink/bin/flink run  /export/server/flink/examples/batch/WordCount.jar --input hdfs://node1:8020/wordcount/input/words.txt --output hdfs://node1:8020/wordcount/output/result.txt  --parallelism 2

state.backend: filesystem

state.backend.fs.checkpointdir: hdfs://node1:8020/flink-checkpoints

high-availability: zookeeper

high-availability.storageDir: hdfs://node1:8020/flink/ha/

high-availability.zookeeper.quorum: node1:2181,node2:2181,node3:2181

#開啟HA，使用檔案系統作為快照儲存

state.backend: filesystem

#啟用檢查點，可以將快照儲存到HDFS

state.backend.fs.checkpointdir: hdfs://node1:8020/flink-checkpoints

#使用zookeeper搭建高可用

high-availability: zookeeper

# 儲存JobManager的元資料到HDFS

high-availability.storageDir: hdfs://node1:8020/flink/ha/

# 配置ZK叢集地址

high-availability.zookeeper.quorum: node1:2181,node2:2181,node3:2181

node1:8081

node2:8081

scp -r /export/server/flink/conf/flink-conf.yaml node2:/export/server/flink/conf/

scp -r /export/server/flink/conf/flink-conf.yaml node3:/export/server/flink/conf/

scp -r /export/server/flink/conf/masters node2:/export/server/flink/conf/

scp -r /export/server/flink/conf/masters node3:/export/server/flink/conf/

jobmanager.rpc.address: node2

<!-- 關閉yarn記憶體檢查 -->

<property>

<name>yarn.nodemanager.pmem-check-enabled</name>

    <value>false</value>

</property>

<property>

     <name>yarn.nodemanager.vmem-check-enabled</name>

     <value>false</value>

</property>

scp -r /export/server/hadoop/etc/hadoop/yarn-site.xml node2:/export/server/hadoop/etc/hadoop/yarn-site.xml

scp -r /export/server/hadoop/etc/hadoop/yarn-site.xml node3:/export/server/hadoop/etc/hadoop/yarn-site.xml

[root@node1 bin]# /export/server/flink/bin/flink --help

./flink <ACTION> [OPTIONS] [ARGUMENTS]

The following actions are available:

Action "run" compiles and runs a program.

  Syntax: run [OPTIONS] <jar-file> <arguments>

  "run" action options:

     -c,--class <classname>               Class with the program entry point

                                          ("main()" method). Only needed if the

                                          JAR file does not specify the class in

                                          its manifest.

     -C,--classpath <url>                 Adds a URL to each user code

                                          classloader  on all nodes in the

                                          cluster. The paths must specify a

                                          protocol (e.g. file://) and be

                                          accessible on all nodes (e.g. by means

                                          of a NFS share). You can use this

                                          option multiple times for specifying

                                          more than one URL. The protocol must

                                          be supported by the {@link

                                          java.net.URLClassLoader}.

     -d,--detached                        If present, runs the job in detached

                                          mode

     -n,--allowNonRestoredState           Allow to skip savepoint state that

                                          cannot be restored. You need to allow

                                          this if you removed an operator from

                                          your program that was part of the

                                          program when the savepoint was

                                          triggered.

     -p,--parallelism <parallelism>       The parallelism with which to run the

                                          program. Optional flag to override the

                                          default value specified in the

                                          configuration.

     -py,--python <pythonFile>            Python script with the program entry

                                          point. The dependent resources can be

                                          configured with the `--pyFiles`

                                          option.

     -pyarch,--pyArchives <arg>           Add python archive files for job. The

                                          archive files will be extracted to the

                                          working directory of python UDF

                                          worker. Currently only zip-format is

                                          supported. For each archive file, a

                                          target directory be specified. If the

                                          target directory name is specified,

                                          the archive file will be extracted to

                                          a name can directory with the

                                          specified name. Otherwise, the archive

                                          file will be extracted to a directory

                                          with the same name of the archive

                                          file. The files uploaded via this

                                          option are accessible via relative

                                          path. '#' could be used as the

                                          separator of the archive file path and

                                          the target directory name. Comma (',')

                                          could be used as the separator to

                                          specify multiple archive files. This

                                          option can be used to upload the

                                          virtual environment, the data files

                                          used in Python UDF (e.g.: --pyArchives

                                          file:///tmp/py37.zip,file:///tmp/data.

                                          zip#data --pyExecutable

                                          py37.zip/py37/bin/python). The data

                                          files could be accessed in Python UDF,

                                          e.g.: f = open('data/data.txt', 'r').

     -pyexec,--pyExecutable <arg>         Specify the path of the python

                                          interpreter used to execute the python

                                          UDF worker (e.g.: --pyExecutable

                                          /usr/local/bin/python3). The python

                                          UDF worker depends on Python 3.5+,

                                          Apache Beam (version == 2.23.0), Pip

                                          (version >= 7.1.0) and SetupTools

                                          (version >= 37.0.0). Please ensure

                                          that the specified environment meets

                                          the above requirements.

     -pyfs,--pyFiles <pythonFiles>        Attach custom python files for job.

                                          These files will be added to the

                                          PYTHONPATH of both the local client

                                          and the remote python UDF worker. The

                                          standard python resource file suffixes

                                          such as .py/.egg/.zip or directory are

                                          all supported. Comma (',') could be

                                          used as the separator to specify

                                          multiple files (e.g.: --pyFiles

                                          file:///tmp/myresource.zip,hdfs:///$na

                                          menode_address/myresource2.zip).

     -pym,--pyModule <pythonModule>       Python module with the program entry

                                          point. This option must be used in

                                          conjunction with `--pyFiles`.

     -pyreq,--pyRequirements <arg>        Specify a requirements.txt file which

                                          defines the third-party dependencies.

                                          These dependencies will be installed

                                          and added to the PYTHONPATH of the

                                          python UDF worker. A directory which

                                          contains the installation packages of

                                          these dependencies could be specified

                                          optionally. Use '#' as the separator

                                          if the optional parameter exists

                                          (e.g.: --pyRequirements

                                          file:///tmp/requirements.txt#file:///t

                                          mp/cached_dir).

     -s,--fromSavepoint <savepointPath>   Path to a savepoint to restore the job

                                          from (for example

                                          hdfs:///flink/savepoint-1537).

     -sae,--shutdownOnAttachedExit        If the job is submitted in attached

                                          mode, perform a best-effort cluster

                                          shutdown when the CLI is terminated

                                          abruptly, e.g., in response to a user

                                          interrupt, such as typing Ctrl + C.

  Options for Generic CLI mode:

     -D <property=value>   Allows specifying multiple generic configuration

                           options. The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stabl

                           e/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is

                           also available with the "Application Mode".

                           The name of the executor to be used for executing the

                           given job, which is equivalent to the

                           "execution.target" config option. The currently

                           available executors are: "remote", "local",

                           "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application,

                           which is equivalent to the "execution.target" config

                           option. For the "run" action the currently available

                           targets are: "remote", "local", "kubernetes-session",

                           "yarn-per-job", "yarn-session". For the

                           "run-application" action the currently available

                           targets are: "kubernetes-application",

                           "yarn-application".

  Options for yarn-cluster mode:

     -d,--detached                        If present, runs the job in detached

                                          mode

     -m,--jobmanager <arg>                Set to yarn-cluster to use YARN

                                          execution mode.

     -yat,--yarnapplicationType <arg>     Set a custom application type for the

                                          application on YARN

     -yD <property=value>                 use value for given property

     -yd,--yarndetached                   If present, runs the job in detached

                                          mode (deprecated; use non-YARN

                                          specific option instead)

     -yh,--yarnhelp                       Help for the Yarn session CLI.

     -yid,--yarnapplicationId <arg>       Attach to running YARN session

     -yj,--yarnjar <arg>                  Path to Flink jar file

     -yjm,--yarnjobManagerMemory <arg>    Memory for JobManager Container with

                                          optional unit (default: MB)

     -ynl,--yarnnodeLabel <arg>           Specify YARN node label for the YARN

                                          application

     -ynm,--yarnname <arg>                Set a custom name for the application

                                          on YARN

     -yq,--yarnquery                      Display available YARN resources

                                          (memory, cores)

     -yqu,--yarnqueue <arg>               Specify YARN queue.

     -ys,--yarnslots <arg>                Number of slots per TaskManager

     -yt,--yarnship <arg>                 Ship files in the specified directory

                                          (t for transfer)

     -ytm,--yarntaskManagerMemory <arg>   Memory per TaskManager Container with

                                          optional unit (default: MB)

     -yz,--yarnzookeeperNamespace <arg>   Namespace to create the Zookeeper

                                          sub-paths for high availability mode

     -z,--zookeeperNamespace <arg>        Namespace to create the Zookeeper

                                          sub-paths for high availability mode

  Options for default mode:

     -D <property=value>             Allows specifying multiple generic

                                     configuration options. The available

                                     options can be found at

                                     https://ci.apache.org/projects/flink/flink-

                                     docs-stable/ops/config.html

     -m,--jobmanager <arg>           Address of the JobManager to which to

                                     connect. Use this flag to connect to a

                                     different JobManager than the one specified

                                     in the configuration. Attention: This

                                     option is respected only if the

                                     high-availability configuration is NONE.

     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths

                                     for high availability mode

Action "run-application" runs an application in Application Mode.

  Syntax: run-application [OPTIONS] <jar-file> <arguments>

  Options for Generic CLI mode:

     -D <property=value>   Allows specifying multiple generic configuration

                           options. The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stabl

                           e/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is

                           also available with the "Application Mode".

                           The name of the executor to be used for executing the

                           given job, which is equivalent to the

                           "execution.target" config option. The currently

                           available executors are: "remote", "local",

                           "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application,

                           which is equivalent to the "execution.target" config

                           option. For the "run" action the currently available

                           targets are: "remote", "local", "kubernetes-session",

                           "yarn-per-job", "yarn-session". For the

                           "run-application" action the currently available

                           targets are: "kubernetes-application",

                           "yarn-application".

Action "info" shows the optimized execution plan of the program (JSON).

  Syntax: info [OPTIONS] <jar-file> <arguments>

  "info" action options:

     -c,--class <classname>           Class with the program entry point

                                      ("main()" method). Only needed if the JAR

                                      file does not specify the class in its

                                      manifest.

     -p,--parallelism <parallelism>   The parallelism with which to run the

                                      program. Optional flag to override the

                                      default value specified in the

                                      configuration.

Action "list" lists running and scheduled programs.

  Syntax: list [OPTIONS]

  "list" action options:

     -a,--all         Show all programs and their JobIDs

     -r,--running     Show only running programs and their JobIDs

     -s,--scheduled   Show only scheduled programs and their JobIDs

  Options for Generic CLI mode:

     -D <property=value>   Allows specifying multiple generic configuration

                           options. The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stabl

                           e/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is

                           also available with the "Application Mode".

                           The name of the executor to be used for executing the

                           given job, which is equivalent to the

                           "execution.target" config option. The currently

                           available executors are: "remote", "local",

                           "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application,

                           which is equivalent to the "execution.target" config

                           option. For the "run" action the currently available

                           targets are: "remote", "local", "kubernetes-session",

                           "yarn-per-job", "yarn-session". For the

                           "run-application" action the currently available

                           targets are: "kubernetes-application",

                           "yarn-application".

  Options for yarn-cluster mode:

     -m,--jobmanager <arg>            Set to yarn-cluster to use YARN execution

                                      mode.

     -yid,--yarnapplicationId <arg>   Attach to running YARN session

     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper

                                      sub-paths for high availability mode

  Options for default mode:

     -D <property=value>             Allows specifying multiple generic

                                     configuration options. The available

                                     options can be found at

                                     https://ci.apache.org/projects/flink/flink-

                                     docs-stable/ops/config.html

     -m,--jobmanager <arg>           Address of the JobManager to which to

                                     connect. Use this flag to connect to a

                                     different JobManager than the one specified

                                     in the configuration. Attention: This

                                     option is respected only if the

                                     high-availability configuration is NONE.

     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths

                                     for high availability mode

Action "stop" stops a running program with a savepoint (streaming jobs only).

  Syntax: stop [OPTIONS] <Job ID>

  "stop" action options:

     -d,--drain                           Send MAX_WATERMARK before taking the

                                          savepoint and stopping the pipelne.

     -p,--savepointPath <savepointPath>   Path to the savepoint (for example

                                          hdfs:///flink/savepoint-1537). If no

                                          directory is specified, the configured

                                          default will be used

                                          ("state.savepoints.dir").

  Options for Generic CLI mode:

     -D <property=value>   Allows specifying multiple generic configuration

                           options. The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stabl

                           e/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is

                           also available with the "Application Mode".

                           The name of the executor to be used for executing the

                           given job, which is equivalent to the

                           "execution.target" config option. The currently

                           available executors are: "remote", "local",

                           "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application,

                           which is equivalent to the "execution.target" config

                           option. For the "run" action the currently available

                           targets are: "remote", "local", "kubernetes-session",

                           "yarn-per-job", "yarn-session". For the

                           "run-application" action the currently available

                           targets are: "kubernetes-application",

                           "yarn-application".

  Options for yarn-cluster mode:

     -m,--jobmanager <arg>            Set to yarn-cluster to use YARN execution

                                      mode.

     -yid,--yarnapplicationId <arg>   Attach to running YARN session

     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper

                                      sub-paths for high availability mode

  Options for default mode:

     -D <property=value>             Allows specifying multiple generic

                                     configuration options. The available

                                     options can be found at

                                     https://ci.apache.org/projects/flink/flink-

                                     docs-stable/ops/config.html

     -m,--jobmanager <arg>           Address of the JobManager to which to

                                     connect. Use this flag to connect to a

                                     different JobManager than the one specified

                                     in the configuration. Attention: This

                                     option is respected only if the

                                     high-availability configuration is NONE.

     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths

                                     for high availability mode

Action "cancel" cancels a running program.

  Syntax: cancel [OPTIONS] <Job ID>

  "cancel" action options:

     -s,--withSavepoint <targetDirectory>   **DEPRECATION WARNING**: Cancelling

                                            a job with savepoint is deprecated.

                                            Use "stop" instead.

                                            Trigger savepoint and cancel job.

                                            The target directory is optional. If

                                            no directory is specified, the

                                            configured default directory

                                            (state.savepoints.dir) is used.

  Options for Generic CLI mode:

     -D <property=value>   Allows specifying multiple generic configuration

                           options. The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stabl

                           e/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is

                           also available with the "Application Mode".

                           The name of the executor to be used for executing the

                           given job, which is equivalent to the

                           "execution.target" config option. The currently

                           available executors are: "remote", "local",

                           "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application,

                           which is equivalent to the "execution.target" config

                           option. For the "run" action the currently available

                           targets are: "remote", "local", "kubernetes-session",

                           "yarn-per-job", "yarn-session". For the

                           "run-application" action the currently available

                           targets are: "kubernetes-application",

                           "yarn-application".

  Options for yarn-cluster mode:

     -m,--jobmanager <arg>            Set to yarn-cluster to use YARN execution

                                      mode.

     -yid,--yarnapplicationId <arg>   Attach to running YARN session

     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper

                                      sub-paths for high availability mode

  Options for default mode:

     -D <property=value>             Allows specifying multiple generic

                                     configuration options. The available

                                     options can be found at

                                     https://ci.apache.org/projects/flink/flink-

                                     docs-stable/ops/config.html

     -m,--jobmanager <arg>           Address of the JobManager to which to

                                     connect. Use this flag to connect to a

                                     different JobManager than the one specified

                                     in the configuration. Attention: This

                                     option is respected only if the

                                     high-availability configuration is NONE.

     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths

                                     for high availability mode

Action "savepoint" triggers savepoints for a running job or disposes existing ones.

  Syntax: savepoint [OPTIONS] <Job ID> [<target directory>]

  "savepoint" action options:

     -d,--dispose <arg>       Path of savepoint to dispose.

     -j,--jarfile <jarfile>   Flink program JAR file.

  Options for Generic CLI mode:

     -D <property=value>   Allows specifying multiple generic configuration

                           options. The available options can be found at

                           https://ci.apache.org/projects/flink/flink-docs-stabl

                           e/ops/config.html

     -e,--executor <arg>   DEPRECATED: Please use the -t option instead which is

                           also available with the "Application Mode".

                           The name of the executor to be used for executing the

                           given job, which is equivalent to the

                           "execution.target" config option. The currently

                           available executors are: "remote", "local",

                           "kubernetes-session", "yarn-per-job", "yarn-session".

     -t,--target <arg>     The deployment target for the given application,

                           which is equivalent to the "execution.target" config

                           option. For the "run" action the currently available

                           targets are: "remote", "local", "kubernetes-session",

                           "yarn-per-job", "yarn-session". For the

                           "run-application" action the currently available

                           targets are: "kubernetes-application",

                           "yarn-application".

  Options for yarn-cluster mode:

     -m,--jobmanager <arg>            Set to yarn-cluster to use YARN execution

                                      mode.

     -yid,--yarnapplicationId <arg>   Attach to running YARN session

     -z,--zookeeperNamespace <arg>    Namespace to create the Zookeeper

                                      sub-paths for high availability mode

  Options for default mode:

     -D <property=value>             Allows specifying multiple generic

                                     configuration options. The available

                                     options can be found at

                                     https://ci.apache.org/projects/flink/flink-

                                     docs-stable/ops/config.html

     -m,--jobmanager <arg>           Address of the JobManager to which to

                                     connect. Use this flag to connect to a

                                     different JobManager than the one specified

                                     in the configuration. Attention: This

                                     option is respected only if the

                                     high-availability configuration is NONE.

     -z,--zookeeperNamespace <arg>   Namespace to create the Zookeeper sub-paths

                                     for high availability mode

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>cn.itcast</groupId>
    <artifactId>flink_study_42</artifactId>
    <version>1.0-SNAPSHOT</version>

    <!-- 指定倉庫位置，依次為aliyun、apache和cloudera倉庫 -->
    <repositories>
        <repository>
            <id>aliyun</id>
            <url>http://maven.aliyun.com/nexus/content/groups/public/</url>
        </repository>
        <repository>
            <id>apache</id>
            <url>https://repository.apache.org/content/repositories/snapshots/</url>
        </repository>
        <repository>
            <id>cloudera</id>
            <url>https://repository.cloudera.com/artifactory/cloudera-repos/</url>
        </repository>
    </repositories>

    <properties>
        <encoding>UTF-8</encoding>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <java.version>1.8</java.version>
        <scala.version>2.12</scala.version>
        <flink.version>1.12.0</flink.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- flink執行計劃,這是1.9版本之前的-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <!-- blink執行計劃,1.11+預設的-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!--<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-cep_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>-->

        <!-- flink聯結器-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-kafka_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-csv</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- <dependency>
           <groupId>org.apache.flink</groupId>
           <artifactId>flink-connector-filesystem_2.12</artifactId>
           <version>${flink.version}</version>
       </dependency>-->
        <!--<dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-jdbc_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>-->
        <!--<dependency>
              <groupId>org.apache.flink</groupId>
              <artifactId>flink-parquet_2.12</artifactId>
              <version>${flink.version}</version>
         </dependency>-->
        <!--<dependency>
            <groupId>org.apache.avro</groupId>
            <artifactId>avro</artifactId>
            <version>1.9.2</version>
        </dependency>
        <dependency>
            <groupId>org.apache.parquet</groupId>
            <artifactId>parquet-avro</artifactId>
            <version>1.10.0</version>
        </dependency>-->


        <dependency>
            <groupId>org.apache.bahir</groupId>
            <artifactId>flink-connector-redis_2.11</artifactId>
            <version>1.0</version>
            <exclusions>
                <exclusion>
                    <artifactId>flink-streaming-java_2.11</artifactId>
                    <groupId>org.apache.flink</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>flink-runtime_2.11</artifactId>
                    <groupId>org.apache.flink</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>flink-core</artifactId>
                    <groupId>org.apache.flink</groupId>
                </exclusion>
                <exclusion>
                    <artifactId>flink-java</artifactId>
                    <groupId>org.apache.flink</groupId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-hive_2.12</artifactId>
            <version>${flink.version}</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-metastore</artifactId>
            <version>2.1.0</version>
        </dependency>
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-exec</artifactId>
            <version>2.1.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-shaded-hadoop-2-uber</artifactId>
            <version>2.7.5-10.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.hbase</groupId>
            <artifactId>hbase-client</artifactId>
            <version>2.1.0</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.38</version>
            <!--<version>8.0.20</version>-->
        </dependency>

        <!-- 高效能非同步元件：Vertx-->
        <dependency>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-core</artifactId>
            <version>3.9.0</version>
        </dependency>
        <dependency>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-jdbc-client</artifactId>
            <version>3.9.0</version>
        </dependency>
        <dependency>
            <groupId>io.vertx</groupId>
            <artifactId>vertx-redis-client</artifactId>
            <version>3.9.0</version>
        </dependency>

        <!-- 日誌 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>1.7.7</version>
            <scope>runtime</scope>
        </dependency>
        <dependency>
            <groupId>log4j</groupId>
            <artifactId>log4j</artifactId>
            <version>1.2.17</version>
            <scope>runtime</scope>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.44</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.18.2</version>
            <scope>provided</scope>
        </dependency>

        <!-- 參考：https://blog.csdn.net/f641385712/article/details/84109098-->
        <!--<dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-collections4</artifactId>
            <version>4.4</version>
        </dependency>-->
        <!--<dependency>
            <groupId>org.apache.thrift</groupId>
            <artifactId>libfb303</artifactId>
            <version>0.9.3</version>
            <type>pom</type>
            <scope>provided</scope>
         </dependency>-->
        <!--<dependency>
           <groupId>com.google.guava</groupId>
           <artifactId>guava</artifactId>
           <version>28.2-jre</version>
       </dependency>-->

    </dependencies>

    <build>
        <sourceDirectory>src/main/java</sourceDirectory>
        <plugins>
            <!-- 編譯外掛 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.5.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                    <!--<encoding>${project.build.sourceEncoding}</encoding>-->
                </configuration>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-surefire-plugin</artifactId>
                <version>2.18.1</version>
                <configuration>
                    <useFile>false</useFile>
                    <disableXmlReport>true</disableXmlReport>
                    <includes>
                        <include>**/*Test.*</include>
                        <include>**/*Suite.*</include>
                    </includes>
                </configuration>
            </plugin>
            <!-- 打包外掛(會包含所有依賴) -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <!--
                                        zip -d learn_spark.jar META-INF/*.RSA META-INF/*.DSA META-INF/*.SF -->
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <!-- 設定jar包的入口類(可選) -->
                                    <mainClass></mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

log4j.rootLogger=WARN, console
log4j.appender.console=org.apache.log4j.ConsoleAppender
log4j.appender.console.layout=org.apache.log4j.PatternLayout
log4j.appender.console.layout.ConversionPattern=%d{HH:mm:ss,SSS} %-5p %-60c %x - %m%n

getExecutionEnvironment() //推薦使用

createLocalEnvironment()

createRemoteEnvironment(String host, int port, String... jarFiles)

package cn.itcast.hello;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.UnsortedGrouping;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Desc
 * 需求:使用Flink完成WordCount-DataSet
 * 編碼步驟
 * 1.準備環境-env
 * 2.準備資料-source
 * 3.處理資料-transformation
 * 4.輸出結果-sink
 * 5.觸發執行-execute//如果有print,DataSet不需要呼叫execute,DataStream需要呼叫execute
 */
public class WordCount1 {
    public static void main(String[] args) throws Exception {
        //老版本的批處理API如下,但已經不推薦使用了
        //1.準備環境-env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        //2.準備資料-source
        DataSet<String> lineDS = env.fromElements("itcast hadoop spark","itcast hadoop spark","itcast hadoop","itcast");
        //3.處理資料-transformation
        //3.1每一行資料按照空格切分成一個個的單片語成一個集合
        /*
        public interface FlatMapFunction<T, O> extends Function, Serializable {
            void flatMap(T value, Collector<O> out) throws Exception;
        }
         */
        DataSet<String> wordsDS = lineDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value就是一行行的資料
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);//將切割處理的一個個的單詞收集起來並返回
                }
            }
        });
        //3.2對集合中的每個單詞記為1
        /*
        public interface MapFunction<T, O> extends Function, Serializable {
            O map(T value) throws Exception;
        }
         */
        DataSet<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是進來一個個的單詞
                return Tuple2.of(value, 1);
            }
        });

        //3.3對資料按照單詞(key)進行分組
        //0表示按照tuple中的索引為0的欄位,也就是key(單詞)進行分組
        UnsortedGrouping<Tuple2<String, Integer>> groupedDS = wordAndOnesDS.groupBy(0);

        //3.4對各個組內的資料按照數量(value)進行聚合就是求sum
        //1表示按照tuple中的索引為1的欄位也就是按照數量進行聚合累加!
        DataSet<Tuple2<String, Integer>> aggResult = groupedDS.sum(1);

        //3.5排序
        DataSet<Tuple2<String, Integer>> result = aggResult.sortPartition(1, Order.DESCENDING).setParallelism(1);

        //4.輸出結果-sink
        result.print();

        //5.觸發執行-execute//如果有print,DataSet不需要呼叫execute,DataStream需要呼叫execute
        //env.execute();//'execute()', 'count()', 'collect()', or 'print()'.
    }
}

package cn.itcast.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Desc
 * 需求:使用Flink完成WordCount-DataStream
 * 編碼步驟
 * 1.準備環境-env
 * 2.準備資料-source
 * 3.處理資料-transformation
 * 4.輸出結果-sink
 * 5.觸發執行-execute
 */
public class WordCount2 {
    public static void main(String[] args) throws Exception {
        //新版本的流批統一API,既支援流處理也支援批處理
        //1.準備環境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2.準備資料-source
        DataStream<String> linesDS = env.fromElements("itcast hadoop spark","itcast hadoop spark","itcast hadoop","itcast");

        //3.處理資料-transformation
        //3.1每一行資料按照空格切分成一個個的單片語成一個集合
        /*
        public interface FlatMapFunction<T, O> extends Function, Serializable {
            void flatMap(T value, Collector<O> out) throws Exception;
        }
         */
        DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value就是一行行的資料
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);//將切割處理的一個個的單詞收集起來並返回
                }
            }
        });
        //3.2對集合中的每個單詞記為1
        /*
        public interface MapFunction<T, O> extends Function, Serializable {
            O map(T value) throws Exception;
        }
         */
        DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是進來一個個的單詞
                return Tuple2.of(value, 1);
            }
        });

        //3.3對資料按照單詞(key)進行分組
        //0表示按照tuple中的索引為0的欄位,也就是key(單詞)進行分組
        //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);

        //3.4對各個組內的資料按照數量(value)進行聚合就是求sum
        //1表示按照tuple中的索引為1的欄位也就是按照數量進行聚合累加!
        DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.輸出結果-sink
        result.print();

        //5.觸發執行-execute
        env.execute();//DataStream需要呼叫execute
    }
}

package cn.itcast.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * Author itcast
 * Desc
 * 需求:使用Flink完成WordCount-DataStream--使用lambda表示式
 * 編碼步驟
 * 1.準備環境-env
 * 2.準備資料-source
 * 3.處理資料-transformation
 * 4.輸出結果-sink
 * 5.觸發執行-execute
 */
public class WordCount3_Lambda {
    public static void main(String[] args) throws Exception {
        //1.準備環境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2.準備資料-source
        DataStream<String> linesDS = env.fromElements("itcast hadoop spark", "itcast hadoop spark", "itcast hadoop", "itcast");
        //3.處理資料-transformation
        //3.1每一行資料按照空格切分成一個個的單片語成一個集合
        /*
        public interface FlatMapFunction<T, O> extends Function, Serializable {
            void flatMap(T value, Collector<O> out) throws Exception;
        }
         */
        //lambda表示式的語法:
        // (引數)->{方法體/函式體}
        //lambda表示式就是一個函式,函式的本質就是物件
        DataStream<String> wordsDS = linesDS.flatMap(
                (String value, Collector<String> out) -> Arrays.stream(value.split(" ")).forEach(out::collect)
        ).returns(Types.STRING);

        //3.2對集合中的每個單詞記為1
        /*
        public interface MapFunction<T, O> extends Function, Serializable {
            O map(T value) throws Exception;
        }
         */
        /*DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(
                (String value) -> Tuple2.of(value, 1)
        ).returns(Types.TUPLE(Types.STRING, Types.INT));*/
        DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(
                (String value) -> Tuple2.of(value, 1)
                , TypeInformation.of(new TypeHint<Tuple2<String, Integer>>() {})
        );

        //3.3對資料按照單詞(key)進行分組
        //0表示按照tuple中的索引為0的欄位,也就是key(單詞)進行分組
        //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
        //KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy((KeySelector<Tuple2<String, Integer>, String>) t -> t.f0);
        KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);

        //3.4對各個組內的資料按照數量(value)進行聚合就是求sum
        //1表示按照tuple中的索引為1的欄位也就是按照數量進行聚合累加!
        DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.輸出結果-sink
        result.print();

        //5.觸發執行-execute
        env.execute();
    }
}

package cn.itcast.hello;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.Arrays;

/**
 * Author itcast
 * Desc
 * 需求:使用Flink完成WordCount-DataStream--使用lambda表示式--修改程式碼使適合在Yarn上執行
 * 編碼步驟
 * 1.準備環境-env
 * 2.準備資料-source
 * 3.處理資料-transformation
 * 4.輸出結果-sink
 * 5.觸發執行-execute//批處理不需要呼叫!流處理需要
 */
public class WordCount4_Yarn {
    public static void main(String[] args) throws Exception {
        //獲取引數
        ParameterTool params = ParameterTool.fromArgs(args);
        String output = null;
        if (params.has("output")) {
            output = params.get("output");
        } else {
            output = "hdfs://node1:8020/wordcount/output_" + System.currentTimeMillis();
        }

        //1.準備環境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //env.setRuntimeMode(RuntimeExecutionMode.STREAMING);
        //env.setRuntimeMode(RuntimeExecutionMode.BATCH);

        //2.準備資料-source
        DataStream<String> linesDS = env.fromElements("itcast hadoop spark", "itcast hadoop spark", "itcast hadoop", "itcast");
        //3.處理資料-transformation
        DataStream<Tuple2<String, Integer>> result = linesDS
                .flatMap(
                        (String value, Collector<String> out) -> Arrays.stream(value.split(" ")).forEach(out::collect)
                ).returns(Types.STRING)
                .map(
                        (String value) -> Tuple2.of(value, 1)
                ).returns(Types.TUPLE(Types.STRING, Types.INT))
                //.keyBy(0);
                .keyBy((KeySelector<Tuple2<String, Integer>, String>) t -> t.f0)
                .sum(1);

        //4.輸出結果-sink
        result.print();

        //如果執行報hdfs許可權相關錯誤,可以執行 hadoop fs -chmod -R 777  /
        System.setProperty("HADOOP_USER_NAME", "root");//設定使用者名稱
        //result.writeAsText("hdfs://node1:8020/wordcount/output_"+System.currentTimeMillis()).setParallelism(1);
        result.writeAsText(output).setParallelism(1);

        //5.觸發執行-execute
        env.execute();
    }
}

package cn.itcast.source;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.Arrays;

/**
 * Author itcast
 * Desc
 * 把本地的普通的Java集合/Scala集合變為分散式的Flink的DataStream集合!
 * 一般用於學習測試時編造資料時使用
 * 1.env.fromElements(可變引數);
 * 2.env.fromColletion(各種集合);
 * 3.env.generateSequence(開始,結束);
 * 4.env.fromSequence(開始,結束);
 */
public class SourceDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //2.source
        // * 1.env.fromElements(可變引數);
        DataStream<String> ds1 = env.fromElements("hadoop", "spark", "flink");
        // * 2.env.fromColletion(各種集合);
        DataStream<String> ds2 = env.fromCollection(Arrays.asList("hadoop", "spark", "flink"));
        // * 3.env.generateSequence(開始,結束);
        DataStream<Long> ds3 = env.generateSequence(1, 10);
        //* 4.env.fromSequence(開始,結束);
        DataStream<Long> ds4 = env.fromSequence(1, 10);
        //3.Transformation
        //4.sink
        ds1.print();
        ds2.print();
        ds3.print();
        ds4.print();
        //5.execute
        env.execute();
    }
}

package cn.itcast.source;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Desc
 * 1.env.readTextFile(本地/HDFS檔案/資料夾);//壓縮檔案也可以
 */
public class SourceDemo02 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //2.source
        // * 1.env.readTextFile(本地檔案/HDFS檔案);//壓縮檔案也可以
        DataStream<String> ds1 = env.readTextFile("data/input/words.txt");
        DataStream<String> ds2 = env.readTextFile("data/input/dir");
        DataStream<String> ds3 = env.readTextFile("hdfs://node1:8020//wordcount/input/words.txt");
        DataStream<String> ds4 = env.readTextFile("data/input/wordcount.txt.gz");
        //3.Transformation
        //4.sink
        ds1.print();
        ds2.print();
        ds3.print();
        ds4.print();
        //5.execute
        env.execute();
    }
}

package cn.itcast.source;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Desc
 * SocketSource
 */
public class SourceDemo03 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //2.source
        DataStream<String> linesDS = env.socketTextStream("node1", 9999);

        //3.處理資料-transformation
        //3.1每一行資料按照空格切分成一個個的單片語成一個集合
        DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value就是一行行的資料
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);//將切割處理的一個個的單詞收集起來並返回
                }
            }
        });
        //3.2對集合中的每個單詞記為1
        DataStream<Tuple2<String, Integer>> wordAndOnesDS = wordsDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是進來一個個的單詞
                return Tuple2.of(value, 1);
            }
        });

        //3.3對資料按照單詞(key)進行分組
        //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
        KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);
        //3.4對各個組內的資料按照數量(value)進行聚合就是求sum
        DataStream<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.輸出結果-sink
        result.print();

        //5.觸發執行-execute
        env.execute();
    }
}

package cn.itcast.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;
import java.util.UUID;

/**
 * Author itcast
 * Desc
 *需求
 * 每隔1秒隨機生成一條訂單資訊(訂單ID、使用者ID、訂單金額、時間戳)
 * 要求:
 * - 隨機生成訂單ID(UUID)
 * - 隨機生成使用者ID(0-2)
 * - 隨機生成訂單金額(0-100)
 * - 時間戳為當前系統時間
 *
 * API
 * 一般用於學習測試,模擬生成一些資料
 * Flink還提供了資料來源介面,我們實現該介面就可以實現自定義資料來源，不同的介面有不同的功能，分類如下：
 * SourceFunction:非並行資料來源(並行度只能=1)
 * RichSourceFunction:多功能非並行資料來源(並行度只能=1)
 * ParallelSourceFunction:並行資料來源(並行度能夠>=1)
 * RichParallelSourceFunction:多功能並行資料來源(並行度能夠>=1)--後續學習的Kafka資料來源使用的就是該介面
 */
public class SourceDemo04_Customer {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //2.Source
        DataStream<Order> orderDS = env
                .addSource(new MyOrderSource())
                .setParallelism(2);

        //3.Transformation

        //4.Sink
        orderDS.print();
        //5.execute
        env.execute();
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Order {
        private String id;
        private Integer userId;
        private Integer money;
        private Long createTime;
    }
    public static class MyOrderSource extends RichParallelSourceFunction<Order> {
        private Boolean flag = true;
        @Override
        public void run(SourceContext<Order> ctx) throws Exception {
            Random random = new Random();
            while (flag){
                Thread.sleep(1000);
                String id = UUID.randomUUID().toString();
                int userId = random.nextInt(3);
                int money = random.nextInt(101);
                long createTime = System.currentTimeMillis();
                ctx.collect(new Order(id,userId,money,createTime));
            }
        }
        //取消任務/執行cancle命令的時候執行
        @Override
        public void cancel() {
            flag = false;
        }
    }
}

CREATE TABLE `t_student` (

    `id` int(11) NOT NULL AUTO_INCREMENT,

    `name` varchar(255) DEFAULT NULL,

    `age` int(11) DEFAULT NULL,

    PRIMARY KEY (`id`)

) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8;

INSERT INTO `t_student` VALUES ('1', 'jack', '18');

INSERT INTO `t_student` VALUES ('2', 'tom', '19');

INSERT INTO `t_student` VALUES ('3', 'rose', '20');

INSERT INTO `t_student` VALUES ('4', 'tom', '19');

INSERT INTO `t_student` VALUES ('5', 'jack', '18');

INSERT INTO `t_student` VALUES ('6', 'rose', '20');

package cn.itcast.source;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.concurrent.TimeUnit;

/**
 * Author itcast
 * Desc
 * 需求:
 * 實際開發中,經常會實時接收一些資料,要和MySQL中儲存的一些規則進行匹配,那麼這時候就可以使用Flink自定義資料來源從MySQL中讀取資料
 * 那麼現在先完成一個簡單的需求:
 * 從MySQL中實時載入資料
 * 要求MySQL中的資料有變化,也能被實時加載出來
 */
public class SourceDemo05_Customer_MySQL {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStream<Student> studentDS = env.addSource(new MySQLSource()).setParallelism(1);

        //3.Transformation
        //4.Sink
        studentDS.print();

        //5.execute
        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    public static class MySQLSource extends RichParallelSourceFunction<Student> {
        private Connection conn = null;
        private PreparedStatement ps = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            //載入驅動,開啟連線
            //Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata", "root", "root");
            String sql = "select id,name,age from t_student";
            ps = conn.prepareStatement(sql);
        }

        private boolean flag = true;

        @Override
        public void run(SourceContext<Student> ctx) throws Exception {
            while (flag) {
                ResultSet rs = ps.executeQuery();
                while (rs.next()) {
                    int id = rs.getInt("id");
                    String name = rs.getString("name");
                    int age = rs.getInt("age");
                    ctx.collect(new Student(id, name, age));
                }
                TimeUnit.SECONDS.sleep(5);
            }
        }
        @Override
        public void cancel() {
            flag = false;
        }
        @Override
        public void close() throws Exception {
            if (conn != null) conn.close();
            if (ps != null) ps.close();
        }
    }
}

package cn.itcast.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Desc
 */
public class TransformationDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //2.source
        DataStream<String> linesDS = env.socketTextStream("node1", 9999);

        //3.處理資料-transformation
        DataStream<String> wordsDS = linesDS.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public void flatMap(String value, Collector<String> out) throws Exception {
                //value就是一行行的資料
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(word);//將切割處理的一個個的單詞收集起來並返回
                }
            }
        });
        DataStream<String> filtedDS = wordsDS.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return !value.equals("heihei");
            }
        });
        DataStream<Tuple2<String, Integer>> wordAndOnesDS = filtedDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                //value就是進來一個個的單詞
                return Tuple2.of(value, 1);
            }
        });
        //KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOnesDS.keyBy(0);
        KeyedStream<Tuple2<String, Integer>, String> groupedDS = wordAndOnesDS.keyBy(t -> t.f0);

        DataStream<Tuple2<String, Integer>> result1 = groupedDS.sum(1);
        DataStream<Tuple2<String, Integer>> result2 = groupedDS.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                return Tuple2.of(value1.f0, value1.f1 + value1.f1);
            }
        });

        //4.輸出結果-sink
        result1.print("result1");
        result2.print("result2");

        //5.觸發執行-execute
        env.execute();
    }
}

package cn.itcast.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.datastream.ConnectedStreams;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.CoMapFunction;

/**
 * Author itcast
 * Desc
 */
public class TransformationDemo02 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //2.Source
        DataStream<String> ds1 = env.fromElements("hadoop", "spark", "flink");
        DataStream<String> ds2 = env.fromElements("hadoop", "spark", "flink");
        DataStream<Long> ds3 = env.fromElements(1L, 2L, 3L);

        //3.Transformation
        DataStream<String> result1 = ds1.union(ds2);//合併但不去重 https://blog.csdn.net/valada/article/details/104367378
        ConnectedStreams<String, Long> tempResult = ds1.connect(ds3);
        //interface CoMapFunction<IN1, IN2, OUT>
        DataStream<String> result2 = tempResult.map(new CoMapFunction<String, Long, String>() {
            @Override
            public String map1(String value) throws Exception {
                return "String->String:" + value;
            }

            @Override
            public String map2(Long value) throws Exception {
                return "Long->String:" + value.toString();
            }
        });

        //4.Sink
        result1.print();
        result2.print();

        //5.execute
        env.execute();
    }
}

package cn.itcast.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * Author itcast
 * Desc
 */
public class TransformationDemo03 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //2.Source
        DataStreamSource<Integer> ds = env.fromElements(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);

        //3.Transformation
        /*SplitStream<Integer> splitResult = ds.split(new OutputSelector<Integer>() {
            @Override
            public Iterable<String> select(Integer value) {
                //value是進來的數字
                if (value % 2 == 0) {
                    //偶數
                    ArrayList<String> list = new ArrayList<>();
                    list.add("偶數");
                    return list;
                } else {
                    //奇數
                    ArrayList<String> list = new ArrayList<>();
                    list.add("奇數");
                    return list;
                }
            }
        });
        DataStream<Integer> evenResult = splitResult.select("偶數");
        DataStream<Integer> oddResult = splitResult.select("奇數");*/

        //定義兩個輸出標籤
        OutputTag<Integer> tag_even = new OutputTag<Integer>("偶數", TypeInformation.of(Integer.class));
        OutputTag<Integer> tag_odd = new OutputTag<Integer>("奇數"){};
        //對ds中的資料進行處理
        SingleOutputStreamOperator<Integer> tagResult = ds.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, Context ctx, Collector<Integer> out) throws Exception {
                if (value % 2 == 0) {
                    //偶數
                    ctx.output(tag_even, value);
                } else {
                    //奇數
                    ctx.output(tag_odd, value);
                }
            }
        });

        //取出標記好的資料
        DataStream<Integer> evenResult = tagResult.getSideOutput(tag_even);
        DataStream<Integer> oddResult = tagResult.getSideOutput(tag_odd);

        //4.Sink
        evenResult.print("偶數");
        oddResult.print("奇數");

        //5.execute
        env.execute();
    }
}

package cn.itcast.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Desc
 */
public class TransformationDemo04 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC).setParallelism(3);

        //2.source
        DataStream<Long> longDS = env.fromSequence(0, 100);

        //3.Transformation
        //下面的操作相當於將資料隨機分配一下,有可能出現數據傾斜
        DataStream<Long> filterDS = longDS.filter(new FilterFunction<Long>() {
            @Override
            public boolean filter(Long num) throws Exception {
                return num > 10;
            }
        });

        //接下來使用map操作,將資料轉為(分割槽編號/子任務編號, 資料)
        //Rich表示多功能的,比MapFunction要多一些API可以供我們使用
        DataStream<Tuple2<Integer, Integer>> result1 = filterDS
                .map(new RichMapFunction<Long, Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> map(Long value) throws Exception {
                        //獲取分割槽編號/子任務編號
                        int id = getRuntimeContext().getIndexOfThisSubtask();
                        return Tuple2.of(id, 1);
                    }
                }).keyBy(t -> t.f0).sum(1);

        DataStream<Tuple2<Integer, Integer>> result2 = filterDS.rebalance()
                .map(new RichMapFunction<Long, Tuple2<Integer, Integer>>() {
                    @Override
                    public Tuple2<Integer, Integer> map(Long value) throws Exception {
                        //獲取分割槽編號/子任務編號
                        int id = getRuntimeContext().getIndexOfThisSubtask();
                        return Tuple2.of(id, 1);
                    }
                }).keyBy(t -> t.f0).sum(1);

        //4.sink
        //result1.print();//有可能出現數據傾斜
        result2.print();//在輸出前進行了rebalance重分割槽平衡,解決了資料傾斜

        //5.execute
        env.execute();
    }
}

package cn.itcast.transformation;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.Partitioner;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Desc
 */
public class TransformationDemo05 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        //2.Source
        DataStream<String> linesDS = env.readTextFile("data/input/words.txt");
        SingleOutputStreamOperator<Tuple2<String, Integer>> tupleDS = linesDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });

        //3.Transformation
        DataStream<Tuple2<String, Integer>> result1 = tupleDS.global();
        DataStream<Tuple2<String, Integer>> result2 = tupleDS.broadcast();
        DataStream<Tuple2<String, Integer>> result3 = tupleDS.forward();
        DataStream<Tuple2<String, Integer>> result4 = tupleDS.shuffle();
        DataStream<Tuple2<String, Integer>> result5 = tupleDS.rebalance();
        DataStream<Tuple2<String, Integer>> result6 = tupleDS.rescale();
        DataStream<Tuple2<String, Integer>> result7 = tupleDS.partitionCustom(new Partitioner<String>() {
            @Override
            public int partition(String key, int numPartitions) {
                return key.equals("hello") ? 0 : 1;
            }
        }, t -> t.f0);

        //4.sink
        //result1.print();
        //result2.print();
        //result3.print();
        //result4.print();
        //result5.print();
        //result6.print();
        result7.print();

        //5.execute
        env.execute();
    }
}

package cn.itcast.sink;

import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Desc
 * 1.ds.print 直接輸出到控制檯
 * 2.ds.printToErr() 直接輸出到控制檯,用紅色
 * 3.ds.collect 將分散式資料收集為本地集合
 * 4.ds.setParallelism(1).writeAsText("本地/HDFS的path",WriteMode.OVERWRITE)
 */
public class SinkDemo01 {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //2.source
        //DataStream<String> ds = env.fromElements("hadoop", "flink");
        DataStream<String> ds = env.readTextFile("data/input/words.txt");

        //3.transformation
        //4.sink
        ds.print();
        ds.printToErr();
        ds.writeAsText("data/output/test", FileSystem.WriteMode.OVERWRITE).setParallelism(2);
        //注意:
        //Parallelism=1為檔案
        //Parallelism>1為資料夾

        //5.execute
        env.execute();
    }
}

package cn.itcast.sink;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Author itcast
 * Desc
 * 使用自定義sink將資料儲存到MySQL
 */
public class SinkDemo02_MySQL {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStream<Student> studentDS = env.fromElements(new Student(null, "tonyma", 18));
        //3.Transformation
        //4.Sink
        studentDS.addSink(new MySQLSink());

        //5.execute
        env.execute();
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }

    public static class MySQLSink extends RichSinkFunction<Student> {
        private Connection conn = null;
        private PreparedStatement ps = null;

        @Override
        public void open(Configuration parameters) throws Exception {
            //載入驅動,開啟連線
            //Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata", "root", "root");
            String sql = "INSERT INTO `t_student` (`id`, `name`, `age`) VALUES (null, ?, ?)";
            ps = conn.prepareStatement(sql);
        }

        @Override
        public void invoke(Student value, Context context) throws Exception {
            //給ps中的?設定具體值
            ps.setString(1,value.getName());
            ps.setInt(2,value.getAge());
            //執行sql
            ps.executeUpdate();
        }

        @Override
        public void close() throws Exception {
            if (conn != null) conn.close();
            if (ps != null) ps.close();
        }
    }
}

package cn.itcast.connectors;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Author itcast
 * Desc
 */
public class ConnectorsDemo_JDBC {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        env.fromElements(new Student(null, "tonyma", 18))
                //3.Transformation
                //4.Sink
                .addSink(JdbcSink.sink(
                        "INSERT INTO `t_student` (`id`, `name`, `age`) VALUES (null, ?, ?)",
                        (ps, s) -> {
                            ps.setString(1, s.getName());
                            ps.setInt(2, s.getAge());
                        },
                        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                                .withUrl("jdbc:mysql://localhost:3306/bigdata")
                                .withUsername("root")
                                .withPassword("root")
                                .withDriverName("com.mysql.jdbc.Driver")
                                .build()));
        //5.execute
        env.execute();
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }
}

package cn.itcast.connectors;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.util.Properties;

/**
 * Author itcast
 * Desc
 * 需求:使用flink-connector-kafka_2.12中的FlinkKafkaConsumer消費Kafka中的資料做WordCount
 * 需要設定如下引數:
 * 1.訂閱的主題
 * 2.反序列化規則
 * 3.消費者屬性-叢集地址
 * 4.消費者屬性-消費者組id(如果不設定,會有預設的,但是預設的不方便管理)
 * 5.消費者屬性-offset重置規則,如earliest/latest...
 * 6.動態分割槽檢測(當kafka的分割槽數變化/增加時,Flink能夠檢測到!)
 * 7.如果沒有設定Checkpoint,那麼可以設定自動提交offset,後續學習了Checkpoint會把offset隨著做Checkpoint的時候提交到Checkpoint和預設主題中
 */
public class ConnectorsDemo_KafkaConsumer {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        Properties props  = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092");
        props.setProperty("group.id", "flink");
        props.setProperty("auto.offset.reset","latest");
        props.setProperty("flink.partition-discovery.interval-millis","5000");//會開啟一個後臺執行緒每隔5s檢測一下Kafka的分割槽情況
        props.setProperty("enable.auto.commit", "true");
        props.setProperty("auto.commit.interval.ms", "2000");
        //kafkaSource就是KafkaConsumer
        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<>("flink_kafka", new SimpleStringSchema(), props);
        kafkaSource.setStartFromGroupOffsets();//設定從記錄的offset開始消費,如果沒有記錄從auto.offset.reset配置開始消費
        //kafkaSource.setStartFromEarliest();//設定直接從Earliest消費,和auto.offset.reset配置無關
        DataStreamSource<String> kafkaDS = env.addSource(kafkaSource);

        //3.Transformation
        //3.1切割並記為1
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = kafkaDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });
        //3.2分組
        KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOneDS.keyBy(0);
        //3.3聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.Sink
        result.print();

        //5.execute
        env.execute();
    }
}

package cn.itcast.connectors;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

/**
 * Author itcast
 * Desc
 * 使用自定義sink-官方提供的flink-connector-kafka_2.12-將資料儲存到Kafka
 */
public class ConnectorsDemo_KafkaProducer {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStreamSource<Student> studentDS = env.fromElements(new Student(1, "tonyma", 18));
        //3.Transformation
        //注意:目前來說我們使用Kafka使用的序列化和反序列化都是直接使用最簡單的字串,所以先將Student轉為字串
        //可以直接呼叫Student的toString,也可以轉為JSON
        SingleOutputStreamOperator<String> jsonDS = studentDS.map(new MapFunction<Student, String>() {
            @Override
            public String map(Student value) throws Exception {
                //String str = value.toString();
                String jsonStr = JSON.toJSONString(value);
                return jsonStr;
            }
        });

        //4.Sink
        jsonDS.print();
        //根據引數建立KafkaProducer/KafkaSink
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "node1:9092");
        FlinkKafkaProducer<String> kafkaSink = new FlinkKafkaProducer<>("flink_kafka",  new SimpleStringSchema(),  props);
        jsonDS.addSink(kafkaSink);

        //5.execute
        env.execute();

        // /export/server/kafka/bin/kafka-console-consumer.sh --bootstrap-server node1:9092 --topic flink_kafka
    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Student {
        private Integer id;
        private String name;
        private Integer age;
    }
}

Data Type

Redis Command [Sink]

HASH

HSET

LIST

RPUSH, LPUSH

SET

SADD

PUBSUB

PUBLISH

STRING

SET

HYPER_LOG_LOG

PFADD

SORTED_SET

ZADD

SORTED_SET

ZREM

package cn.itcast.connectors;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.util.Collector;

/**
 * Author itcast
 * Desc
 * 需求:
 * 接收訊息並做WordCount,
 * 最後將結果儲存到Redis
 * 注意:儲存到Redis的資料結構:使用hash也就是map
 * key            value
 * WordCount    (單詞,數量)
 */
public class ConnectorsDemo_Redis {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2.Source
        DataStream<String> linesDS = env.socketTextStream("node1", 9999);

        //3.Transformation
        //3.1切割並記為1
        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOneDS = linesDS.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] words = value.split(" ");
                for (String word : words) {
                    out.collect(Tuple2.of(word, 1));
                }
            }
        });
        //3.2分組
        KeyedStream<Tuple2<String, Integer>, Tuple> groupedDS = wordAndOneDS.keyBy(0);
        //3.3聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = groupedDS.sum(1);

        //4.Sink
        result.print();
        // * 最後將結果儲存到Redis
        // * 注意:儲存到Redis的資料結構:使用hash也就是map
        // * key            value
        // * WordCount      (單詞,數量)

        //-1.建立RedisSink之前需要建立RedisConfig
        //連線單機版Redis
        FlinkJedisPoolConfig conf = new FlinkJedisPoolConfig.Builder().setHost("127.0.0.1").build();
        //連線叢集版Redis
        //HashSet<InetSocketAddress> nodes = new HashSet<>(Arrays.asList(new InetSocketAddress(InetAddress.getByName("node1"), 6379),new InetSocketAddress(InetAddress.getByName("node2"), 6379),new InetSocketAddress(InetAddress.getByName("node3"), 6379)));
        //FlinkJedisClusterConfig conf2 = new FlinkJedisClusterConfig.Builder().setNodes(nodes).build();
        //連線哨兵版Redis
        //Set<String> sentinels = new HashSet<>(Arrays.asList("node1:26379", "node2:26379", "node3:26379"));
        //FlinkJedisSentinelConfig conf3 = new FlinkJedisSentinelConfig.Builder().setMasterName("mymaster").setSentinels(sentinels).build();

        //-3.建立並使用RedisSink
        result.addSink(new RedisSink<Tuple2<String, Integer>>(conf, new RedisWordCountMapper()));

        //5.execute
        env.execute();
    }

    /**
     * -2.定義一個Mapper用來指定儲存到Redis中的資料結構
     */
    public static class RedisWordCountMapper implements RedisMapper<Tuple2<String, Integer>> {
        @Override
        public RedisCommandDescription getCommandDescription() {
            return new RedisCommandDescription(RedisCommand.HSET, "WordCount");
        }
        @Override
        public String getKeyFromData(Tuple2<String, Integer> data) {
            return data.f0;
        }
        @Override
        public String getValueFromData(Tuple2<String, Integer> data) {
            return data.f1.toString();
        }
    }
}

package cn.itcast.batch;

import org.apache.flink.api.common.JobExecutionResult;
import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;

/**
 * Author itcast
 * Desc 演示Flink累加器,統計處理的資料條數
 */
public class OtherAPI_Accumulator {
    public static void main(String[] args) throws Exception {
        //1.env
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        //2.Source
        DataSource<String> dataDS = env.fromElements("aaa", "bbb", "ccc", "ddd");

        //3.Transformation
        MapOperator<String, String> result = dataDS.map(new RichMapFunction<String, String>() {
            //-1.建立累加器
            private IntCounter elementCounter = new IntCounter();
            Integer count = 0;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //-2註冊累加器
                getRuntimeContext().addAccumulator("elementCounter", elementCounter);
            }

            @Override
            public String map(String value) throws Exception {
                //-3.使用累加器
                this.elementCounter.add(1);
                count+=1;
                System.out.println("不使用累加器統計的結果:"+count);
                return value;
            }
        }).setParallelism(2);

        //4.Sink
        result.writeAsText("data/output/test", FileSystem.WriteMode.OVERWRITE);

        //5.execute
        //-4.獲取加強結果
        JobExecutionResult jobResult = env.execute();
        int nums = jobResult.getAccumulatorResult("elementCounter");
        System.out.println("使用累加器統計的結果:"+nums);
    }
}