全託管Flink DataStream作業

前提條件

已購買Hologres執行個體並建立資料庫。詳情請參見建立資料庫。
已安裝代碼開發平台，用於本地代碼調試，如IntelliJ IDEA。

步驟一：下載Connector依賴

通過DataStream的方式讀寫Hologres資料時，您需要下載Hologres連接器串連Flink全託管。目前發行的連接器版本請參見Hologres DataStream連接器。

您需要下載如下2個依賴JAR包：
- ververica-connector-hologres-1.17-vvr-8.0.8.jar：用於本地調試。
- ververica-connector-hologres-1.17-vvr-8.0.8-uber.jar：用於本地調試和線上部署。
  
  說明
  從VVR-6.0-Flink-1.15版本起，商業版Connector在本地調試時，需要配合相應版本的Uber JAR使用。使用方法請參見本地運行和調試包含連接器的作業。
下載後，使用如下命令將ververica-connector-hologres-1.17-vvr-8.0.8.jar安裝至本地Maven倉庫中：
```
mvn install:install-file -Dfile=$path/ververica-connector-hologres-1.17-vvr-8.0.8.jar -DgroupId=com.alibaba.ververica -DartifactId=ververica-connector-hologres -Dversion=1.17-vvr-8.0.8 -Dpackaging=jar
```
其中$path為您本地存放ververica-connector-hologres-1.17-vvr-8.0.8.jar的絕對路徑。

步驟二：本地開發及調試

您需要在本地完成專案開發，再在Flink全託管控制台上部署並運行。以Binlog源表為例，專案代碼及pom.xml檔案如下：

本地代碼編寫：

DataStream API Demo代碼：

import com.alibaba.ververica.connectors.hologres.binlog.HologresBinlogConfigs;
import com.alibaba.ververica.connectors.hologres.binlog.StartupMode;
import com.alibaba.ververica.connectors.hologres.binlog.source.HologresBinlogSource;
import com.alibaba.ververica.connectors.hologres.config.HologresConnectionParam;
import com.alibaba.ververica.connectors.hologres.config.JDBCOptions;
import com.alibaba.ververica.connectors.hologres.utils.JDBCUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;
import java.util.Collections;
public class HologresBinlogSourceDemo {
    public static void main(String[] args) throws Exception {
        Configuration envConf = new Configuration();
        // 本地調試時，需要指定uber jar的絕對路徑；打包上傳時請注釋掉
        envConf.setString("pipeline.classpaths", "file://" + "<path_to_uber_jar>");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(envConf);
        // 初始化讀取的表的Schema，需要和Hologres表的欄位匹配，可以只定義部分欄位。
        TableSchema schema = TableSchema.builder()
        .field("<id>", DataTypes.INT().notNull())
        .primaryKey("<id>")
        .build();
        // Hologres的相關參數。
        Configuration config = new Configuration();
        config.setString(HologresConfigs.ENDPOINT, "<yourEndpoint>");
        config.setString(HologresConfigs.USERNAME, "<yourUserName>");
        config.setString(HologresConfigs.PASSWORD, "<yourPassword>");
        config.setString(HologresConfigs.DATABASE, "<yourDatabaseName>");
        config.setString(HologresConfigs.TABLE, "<yourTableName>");
        config.setBoolean(HologresBinlogConfigs.OPTIONAL_BINLOG, true);
        config.setBoolean(HologresBinlogConfigs.BINLOG_CDC_MODE, true);
        // 構建JDBC Options。
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // 構建Hologres Binlog Source。
        long startTimeMs = 0;
        HologresBinlogSource source = new HologresBinlogSource(
            new HologresConnectionParam(config),
            schema,
            config,
            jdbcOptions,
            startTimeMs,
            StartupMode.INITIAL,
            "",
            "",
            -1,
            Collections.emptySet());
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Test source").print();
        env.execute();  
    }
}

參數說明：

參數	描述
path_to_uber_jar	本地Uber JAR的絕對路徑。對於Windows需要加相應磁碟分割，例如`file:///D:/path/to/a-uber.jar`。
id	始化讀取的表的Schema，需要和Hologres表的欄位匹配，可以只定義部分欄位。
yourEndpoint	Hologres執行個體的網路網域名稱。您可以進入Hologres管理主控台的執行個體詳情頁，從網絡資訊中擷取網域名稱。
yourUserName	阿里雲帳號的AccessKey ID。您可以單擊AccessKey 管理，擷取AccessKey ID。
yourPassword	對應阿里雲帳號的AccessKey Secret。
yourDatabaseName	Hologres資料庫名稱。
yourTableName	待讀取的Hologres表名稱。

pom.xml檔案：

<?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>com.alibaba.hologres</groupId>
    <artifactId>hologres-flink-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <packaging>jar</packaging>
    <properties>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <flink.version>1.17.2</flink.version>
        <vvr.version>1.17-vvr-8.0.8</vvr.version>
        <target.java.version>1.8</target.java.version>
        <scala.binary.version>2.12</scala.binary.version>
        <maven.compiler.source>${target.java.version}</maven.compiler.source>
        <maven.compiler.target>${target.java.version}</maven.compiler.target>
        <log4j.version>1.7.21</log4j.version>
    </properties>
    <dependencies>
        <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-table-common</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-runtime</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-base</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>
        <dependency>
            <groupId>com.alibaba.ververica</groupId>
            <artifactId>ververica-connector-hologres</artifactId>
            <version>${vvr.version}</version>
        </dependency>
        <!--  日誌實現  log4j  依賴  -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>${log4j.version}</version>
        </dependency>
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-log4j12</artifactId>
            <version>${log4j.version}</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.1.0</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <createDependencyReducedPom>false</createDependencyReducedPom>
                            <shadedArtifactAttached>true</shadedArtifactAttached>
                            <shadedClassifierName>jar-with-dependencies</shadedClassifierName>
                            <transformers>
                                <transformer
                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>
</project>

本地調試及運行。
- 您需要配置運行所需要的ClassLoader JAR包，即ververica-classloader-1.15-vvr-6.0-SNAPSHOT.jar。具體操作請參見步驟二：配置運行所需要的ClassLoader JAR包。
- （可選）如果提示缺少一些常見的Flink類無法執行，例如org.apache.flink.configuration.Configuration，需要在“Modify options”處勾選“Add dependencies with provided scope to classpath”。
配置完成後，您可在本地調試運行該專案，確保本地可運行成功。

本地調試步驟詳情請參見本地運行和調試包含連接器的作業。

步驟三：打包運行

本地調試成功後，您可將其進行打包並與Uber JAR一同上傳至Flink。

打包前，注釋掉下述代碼：

envConf.setString("pipeline.classpaths", "file://" + "<path_to_uber_jar>");

編譯打包。

使用Maven編譯並打包應用程式及其依賴項。命令如下：
```
mvn clean package -DskipTests
```
打包成功後，即可在本地產生名為hologres-flink-demo-1.0-SNAPSHOT-jar-with-dependencies.jar的檔案。
上傳JAR包。

在Flink控制台的資源管理頁面上傳打包好的程式JAR包和ververica-connector-hologres-1.17-vvr-8.0.8-uber.jar。具體操作請參見步驟二：上傳測試JAR包和資料檔案。
部署JAR作業。

在Flink控制台的作業營運頁面部署JAR作業。具體操作及參數資訊請參見步驟三：部署JAR作業。配置部署參數，引擎版本選擇 vvr-8.0.8-flink-1.17，JAR Uri設定為打包產生的 hologres-flink-demo-1.0-SNAPSHOT-jar-with-dependencies.jar 檔案的 OSS 路徑，Entry Point Class填寫 org.example.HologresBinlogSourceDemo，附加依賴檔案添加 ververica-connector-hologres-1.17-vvr-8.0.8-uber.jar 的 OSS 路徑，部署目標選擇 default-queue。
啟動並查看Flink計算結果。

說明
若更新JAR包，需要重新上傳部署JAR包並啟動作業。
1. 在Flink控制台的作業營運頁面，單擊目標作業名稱操作列中的啟動。
2. 配置資源資訊和基礎設定。
  
  作業啟動參數配置詳情請參見作業啟動。
3. 單擊啟動。
  
  單擊啟動後，作業狀態變為运行中，則代表作業運行正常。

常見問題

問題1：當您在IntelliJ IDEA中運行和調試Flink作業時，如果其包含了阿里雲Realtime ComputeFlink版的商業版連接器依賴，可能會遇到無法找到連接器相關類的運行錯誤。例如Caused by: java.lang.ClassNotFoundException: com.alibaba.ververica.connectors.hologres.binlog.source.reader.HologresBinlogRecordEmitter。
- 問題原因：此類異常一般是由於本地調試沒有正確使用uber jar導致的
- 解決方案：請參考本文或者本地運行和調試包含連接器的作業正確使用Uber JAR進行調試。
問題2：提示缺少一些常見的Flink類無法執行，例如Caused by: java.lang.ClassNotFoundException: org.apache.flink.configuration.Configuration。
- 問題原因：可能是缺少依賴或者沒有正常載入依賴。
- 解決方案：
  - pom.xml檔案中沒有引入相關依賴，大多數情況下可能是flink-connector-base，也可以搜尋異常包路徑，查看其屬於哪個Flink依賴。
  - 可能是運行時沒有載入provided依賴。需要在IntelliJ IDEA的“Modify options”處勾選“Add dependencies with provided scope to classpath”。
問題3：運行中報錯Incompatible magic value 。
- 問題原因：
  - 原因一：可能是使用的Uber JAR與Connector版本不一致。
  - 原因二：可能是ClassLoader設定有誤。
- 解決方案：
  - 對於原因一：可參考本文選擇對應版本的Connector和Uber JAR。
  - 對於原因二：請參考配置運行所需要的ClassLoader JAR包重新設定。
問題4：運行時拋出異常Unable to load flink-decryption library java.io.FileNotFoundException: Decryption library native/windows/x86/flink-decryption.dll not found。
- 問題原因：目前Uber JAR不支援Windows系統32位的Java。
- 解決方案：請安裝64位的Java，可以通過java -version命令查看Java安裝資訊，如果不包含64-Bit字樣，表明是32位的Java。
問題5：運行時拋出Caused by: java.lang.ClassFormatError。
- 問題原因：可能是由於IntelliJ IDEA配置的JDK版本問題導致。
- 解決方案：請使用較新的JDK8或者JDK11版本。