Documentation
Home Page Hologres User Guide Data migration Data import Import via Flink Realtime Compute for Apache Flink Fully managed Flink DataStream deployments

Fully managed Flink DataStream deployments

Updated at:
Copy as MD
Product

Hologres is highly compatible with fully managed Flink. In most cases, you can use Flink SQL to declare Hologres source tables, dimension tables, and result tables to define your data processing logic. However, for specific use cases where Flink SQL does not meet your computational needs, you must use the DataStream API to read and write data. This topic walks you through developing and debugging a DataStream deployment that uses the Hologres connector, using engine version VVR-8.0.8-Flink-1.17 as an example.

Prerequisites

  • Purchase a Hologres instance and create a database. For more information, see Create a database.

  • Install an integrated development environment (IDE), such as IntelliJ IDEA, for local debugging.

Step 1: Download the connector dependencies

To read and write data from Hologres using the DataStream API, download the Hologres connector for the fully managed Flink service. For a list of available connector versions, see Hologres DataStream connectors.

  1. Download the following two dependency JARs:

    • ververica-connector-hologres-1.17-vvr-8.0.8.jar: used for local debugging.

    • ververica-connector-hologres-1.17-vvr-8.0.8-uber.jar: used for local debugging and online deployment.

      Note

      Starting from VVR-6.0-Flink-1.15, you must use the corresponding uber JAR when you debug a commercial connector locally. For more information, see Debug connectors locally.

  2. After you download the file, run the following command to install the ververica-connector-hologres-1.17-vvr-8.0.8.jar file into your local Maven repository:

    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

    In the command, replace $path with the absolute path to the ververica-connector-hologres-1.17-vvr-8.0.8.jar file on your local machine.

Step 2: Develop and debug locally

Develop your project locally before you deploy and run it on the fully managed Flink console. The following example shows the project code and pom.xml file for reading from a binary log source table.

  1. Write the project code.

    • DataStream API demo code:

      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();
        // For local debugging, specify the absolute path to the uber JAR. Comment out this line when packaging for upload.
        envConf.setString("pipeline.classpaths", "file://" + "<path_to_uber_jar>");
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(envConf);
        // Initialize the schema of the table to be read. The fields must match the Hologres table, but you can define only a subset of fields.
        TableSchema schema = TableSchema.builder()
        .field("<id>", DataTypes.INT().notNull())
        .primaryKey("<id>")
        .build();
        // Hologres connection parameters.
        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);
        // Build the JDBCOptions.
        JDBCOptions jdbcOptions = JDBCUtils.getJDBCOptions(config);
        // Build the HologresBinlogSource.
        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();  
    }
}

      Parameters:

      Parameter

      Description

      path_to_uber_jar

      The absolute path of the local uber JAR. For Windows, you must include the drive letter, such as file:///D:/path/to/a-uber.jar.

      id

      The name of the primary key column in your Hologres table. This example uses id as a placeholder.

      yourEndpoint

      The endpoint of your Hologres instance. You can find the endpoint in the Network Information section of the instance details page in the Hologres console.

      yourUserName

      The AccessKey ID of your Alibaba Cloud account. You can get it from the AccessKey Management page.

      yourPassword

      The AccessKey secret for your AccessKey ID.

      yourDatabaseName

      The name of your Hologres database.

      yourTableName

      The name of the Hologres table that you want to read.

    • pom.xml file:

      <?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>
        <!-- Logging implementation (log4j) dependencies -->
        <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>

  2. Debug and run the project locally.

    • Configure the ClassLoader JAR required for execution: ververica-classloader-1.15-vvr-6.0-SNAPSHOT.jar. For detailed instructions, see Step 2: Configure the required ClassLoader JAR package.

    • (Optional) If an error occurs because a common Flink class cannot be found, such as org.apache.flink.configuration.Configuration, select the Add dependencies with provided scope to classpath option in your IDE's run configuration.

    After you complete the configuration, debug and run the project locally to ensure it executes successfully.

For detailed instructions on local debugging, see Debug connectors locally.

Step 3: Package and run the deployment

After successfully debugging your project locally, package your application and upload it with the uber JAR to the Flink console.

  1. Before packaging, comment out the following line of code:

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

  2. Compile and package the application.

    Use Maven to compile and package your application and its dependencies. Run the following command:

    mvn clean package -DskipTests

    After the package is successfully created, a file named hologres-flink-demo-1.0-SNAPSHOT-jar-with-dependencies.jar is generated in your local project directory.

  3. Upload the JARs.

    On the Artifacts page of the Realtime Compute for Apache Flink console, upload your packaged application JAR and the ververica-connector-hologres-1.17-vvr-8.0.8-uber.jar file. For detailed instructions, see Step 2: Upload the test JAR package and data file.

  4. Create a JAR deployment.

    On the Deployments page of the Flink console, create a JAR deployment. For detailed instructions and parameter descriptions, see Step 3: Deploy the JAR deployment. Configure the deployment parameters: set Engine Version to vvr-8.0.8-flink-1.17, set JAR URI to the OSS path of your hologres-flink-demo-1.0-SNAPSHOT-jar-with-dependencies.jar file, enter HologresBinlogSourceDemo for Entry Point Class, add the OSS path of ververica-connector-hologres-1.17-vvr-8.0.8-uber.jar to Additional Dependencies, and select default-queue as the Deployment Target.

  5. Start the deployment and view the results.

    Note

    If you update the JAR, you must re-upload it, create a new deployment, and start it.

    1. On the Deployments page of the Flink console, find your deployment and click Start in the Actions column.

    2. Configure the resource information and basic settings.

      For more information about the deployment startup parameters, see Start a deployment.

    3. Click Start.

      After you click Start, the deployment Status changes to Running. A status of RUNNING indicates that the deployment is running correctly.

FAQ

  • Problem: When I run or debug a Flink deployment in IntelliJ IDEA, I get a ClassNotFoundException for a connector-related class, such as Caused by: java.lang.ClassNotFoundException: com.alibaba.ververica.connectors.hologres.binlog.source.reader.HologresBinlogRecordEmitter.

    • Cause: This exception occurs if the uber JAR is not used correctly during local debugging.

    • Solution: Ensure you are using the uber JAR correctly for local debugging. For instructions, see this topic or Debug connectors locally.

  • Problem: I get a ClassNotFoundException for a common Flink class, such as Caused by: java.lang.ClassNotFoundException: org.apache.flink.configuration.Configuration.

    • Cause: A required dependency is missing or was not loaded correctly.

    • Solution:

      • The required dependency is not included in the pom.xml file. In most cases, the missing dependency is flink-connector-base. You can also search for the package path from the error to identify which Flink dependency is missing.

      • The 'provided' scope dependencies were not loaded at runtime. In IntelliJ IDEA, select the Add dependencies with provided scope to classpath option in your run configuration.

  • Problem: The deployment fails with an Incompatible magic value error.

    • Causes:

      • The version of the uber JAR does not match the connector version.

      • The ClassLoader is configured incorrectly.

    • Solutions:

      • To resolve the version mismatch, select the matching connector and uber JAR versions as shown in this topic.

      • To fix the configuration, reconfigure the ClassLoader. For more information, see Configure the required ClassLoader JAR package.

  • Problem: The deployment fails with an Unable to load flink-decryption library java.io.FileNotFoundException: Decryption library native/windows/x86/flink-decryption.dll not found error.

    • Cause: The uber JAR does not support 32-bit Java on Windows.

    • Solution: Install a 64-bit version of Java. You can run the java -version command to check your Java installation. If the output does not contain 64-Bit, you are using a 32-bit version.

  • Problem: The deployment fails with a Caused by: java.lang.ClassFormatError.

    • Cause: This error is caused by an incompatible Java Development Kit (JDK) version configured in IntelliJ IDEA.

    • Solution: Use JDK 8 or JDK 11.

Previous：Realtime Compute for Apache FlinkNext：Import batch data with Flink