Maven環(huán)境配置（可選）

配置Maven的setting.xml文件。如果您在后續(xù)的操作中，對Maven中央倉庫的訪問存在無法拉取或速率較慢的情況，可以更換為阿里云鏡像倉庫。

  <mirror>
      <id>aliyunmaven</id>
        <mirrorOf>central</mirrorOf>
        <name>Aliyun Maven</name>
        <url>https://maven.aliyun.com/repository/public</url>
    </mirror>

配置Flink環(huán)境依賴

下面是Flink的一些基本相關(guān)依賴，您可能還需要補充一些日志文件相關(guān)的依賴，完整的依賴參考請參見文末的完整示例代碼。

         <!-- Apache Flink 依賴項 -->
        <!-- 之所以提供這些依賴項，是因為它們不應(yīng)該打包到JAR文件中。 -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

連接器依賴和使用

通過DataStream的方式讀寫數(shù)據(jù)，需要使用對應(yīng)的DataStream連接器連接實時計算Flink版。Maven中央倉庫已經(jīng)放置了VVR DataStream連接器，以供您在作業(yè)開發(fā)時直接使用。

您可以選擇以下任意一種方式來使用連接器：

OSS附加依賴文件讀取

因為Flink JAR作業(yè)不支持在Main函數(shù)中讀取本地配置，您可以將配置文件上傳到Flink工作空間下的OSS Bucket，在部署JAR作業(yè)時，通過添加附加配置文件的方式進行讀取。示例如下。

1. 創(chuàng)建配置文件config.properties，避免在代碼中出現(xiàn)明文代碼。

   # Kafka 
   bootstrapServers=host1:9092,host2:9092,host3:9092
   inputTopic=topic
   groupId=groupId
   # MySQL
   database.url=jdbc:mysql://localhost:3306/my_database
   database.username=username
   database.password=password

2. 在JAR作業(yè)中使用代碼讀取存儲在OSS Bucket上的配置文件config.properties。

   方式一：讀取工作空間綁定的OSS Bucket

   1. 實時計算開發(fā)控制臺左側(cè)導(dǎo)航欄資源管理頁面，上傳該文件。

   2. 在作業(yè)運行時，部署作業(yè)所添加附加依賴文件將會加載到作業(yè)所運行Pod的/flink/usrlib目錄下。

   3. 讀取該配置文件代碼示例如下。

                  Properties properties = new Properties();
                  Map<String,String> configMap = new HashMap<>();
      
                  try (InputStream input = new FileInputStream("/flink/usrlib/config.properties")) {
                      // 加載屬性文件
                      properties.load(input);
                      // 獲取屬性值
                      configMap.put("bootstrapServers",properties.getProperty("bootstrapServers")) ;
                      configMap.put("inputTopic",properties.getProperty("inputTopic"));
                      configMap.put("groupId",properties.getProperty("groupId"));
                      configMap.put("url",properties.getProperty("database.url")) ;
                      configMap.put("username",properties.getProperty("database.username"));
                      configMap.put("password",properties.getProperty("database.password"));
                  } catch (IOException ex) {
                      ex.printStackTrace();
                  }

   方式二：讀取工作空間有權(quán)限訪問的OSS Bucket

   1. 將配置文件上傳目標(biāo)OSS Bucket。

   2. 通過OSSClient直接讀取OSS上的存儲文件詳情，請參見流式傳輸和管理訪問憑據(jù)。代碼示例如下。

      OSS ossClient = new OSSClientBuilder().build("Endpoint", "AccessKeyId", "AccessKeySecret");
      try (OSSObject ossObject = ossClient.getObject("examplebucket", "exampledir/config.properties");
           BufferedReader reader = new BufferedReader(new InputStreamReader(ossObject.getObjectContent()))) {
          // read file and process ...
      } finally {
          if (ossClient != null) {
              ossClient.shutdown();
          }
      }

業(yè)務(wù)代碼編寫

1. 將外部數(shù)據(jù)源集成到Flink數(shù)據(jù)流程序。Watermark是Flink一種基于時間語義的計算策略，往往伴隨著時間戳一起使用，所以本示例不使用水印策略。詳情請參考水印策略。

            // 將外部數(shù)據(jù)源集成到flink數(shù)據(jù)流程序
           // WatermarkStrategy.noWatermarks() 指沒有使用水印策略
           DataStreamSource<String> stream = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka Source");

2. 算子轉(zhuǎn)換處理。示例中將DataStream<String>轉(zhuǎn)換成DataStream<Student>，更多復(fù)雜的算子轉(zhuǎn)化和處理方式請參考Flink算子。

             // 轉(zhuǎn)換數(shù)據(jù)結(jié)構(gòu)為student的算子
             DataStream<student> source = stream
                   .map(new MapFunction<String, student>() {
                       @Override
                       public student map(String s) throws Exception {
                           // 數(shù)據(jù)由逗號分隔
                           String[] data = s.split(",");
                           return new student(Integer.parseInt(data[0]), data[1], Integer.parseInt(data[2]));
                       }
                   }).filter(student -> student.score >=60); // 篩選出分數(shù)大于60分的數(shù)據(jù)

作業(yè)打包

通過maven-shade-plugin插件打包。

<build>
        <plugins>
            <!-- Java 編譯器-->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>${target.java.version}</source>
                    <target>${target.java.version}</target>
                </configuration>
            </plugin>

            <!-- 我們使用maven-shade-plugin創(chuàng)建一個包含所有必須依賴的fat jar -->
            <!-- 修改<mainClass>的值.如果您的程序入口點發(fā)生了改變 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.2.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <!-- 去掉一些不必要的依賴性 -->
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>org.apache.flink:force-shading</exclude>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>org.apache.logging.log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- 不要復(fù)制META-INF文件夾中的簽名。否則，這可能會在使用JAR文件時導(dǎo)致安全異常 -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <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">
                                    <mainClass>com.aliyun.FlinkDemo</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>