E-MapReduce:FAQ

View cluster logs

Resolve JAR package conflicts

Submit jobs from an external machine

Resolve cluster hostnames from an external machine

Check Flink job status

• Use the EMR console.

  EMR supports Knox, which allows you to access the web UIs of services like YARN and Flink over the internet. You can access the Flink web UI through YARN. For more information, see View the job status on the web UI.

• Use an SSH tunnel. For more information, see Create an SSH tunnel to access web UIs of open source components.

• Access the YARN REST API directly.

  curl --compressed -v  -H "Accept: application/json" -X GET "http://master-1-1:8088/ws/v1/cluster/apps?states=RUNNING&queue=default&user.name=***"

  Note

  Ensure that your security group allows access to ports 8443 and 8088 to reach the YARN REST API. Alternatively, ensure the DataFlow cluster and your client node are in the same Virtual Private Cloud (VPC).

Access Flink job logs

• For a running job, you can access its logs through the Flink web UI.

• For a finished job, you can view its statistics on the Flink HistoryServer or access its logs by running the command yarn logs -applicationId application_xxxx_yyyy. Logs for finished jobs are stored by default in the hdfs:///tmp/logs/$USERNAME/logs/ directory on the HDFS cluster.

Access the Flink HistoryServer

Use commercial connectors

A DataFlow cluster provides many commercial connectors, such as those for Hologres, SLS, MaxCompute, DataHub, Elasticsearch, and ClickHouse. In your Flink jobs, you can use these commercial connectors in addition to open source ones. The following example demonstrates how to use the included Hologres connector.

• Job development

  1. Download the JAR package of the commercial connector from the DataFlow cluster (located in the /opt/apps/FLINK/flink-current/opt/connectors directory). Then, install the connector in your local Maven environment by running the following command.

     mvn install:install-file -Dfile=/path/to/ververica-connector-hologres-1.13-vvr-4.0.7.jar -DgroupId=com.alibaba.ververica -DartifactId=ververica-connector-hologres -Dversion=1.13-vvr-4.0.7 -Dpackaging=jar

  2. Add the following dependency to your project's pom.xml file.

     <dependency>
         <groupId>com.alibaba.ververica</groupId>
         <artifactId>ververica-connector-hologres</artifactId>
         <version>1.13-vvr-4.0.7</version>
         <scope>provided</scope>
     </dependency>

• Run the job

  • Method 1:

    1. Copy the Hologres connector to a separate directory.

       hdfs mkdir hdfs:///flink-current/opt/connectors/hologres/
       hdfs cp hdfs:///flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar  hdfs:///flink-current/opt/connectors/hologres/ververica-connector-hologres-1.13-vvr-4.0.7.jar

    2. When you submit the job, add the following parameter to the command.

       -D yarn.provided.lib.dirs=hdfs:///flink-current/opt/connectors/hologres/

  • Method 2:

    1. Copy the Hologres connector to the /opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar directory on the job submission client. This directory structure must match the one in the DataFlow cluster.

    2. When you submit the job, add the following parameter to the command.

       -C file:///opt/apps/FLINK/flink-current/opt/connectors/ververica-connector-hologres-1.13-vvr-4.0.7.jar

  • Method 3: Package the Hologres connector into your job's JAR package.

Use GeminiStateBackend

A DataFlow cluster provides the enterprise-grade GeminiStateBackend, which offers 3 to 5 times the performance of open source versions. The DataFlow cluster uses GeminiStateBackend by default. For more information about advanced configurations for GeminiStateBackend, see Enterprise-grade state backend configurations.

Use an open source state backend

flink run-application -t yarn-application -D state.backend=rocksdb  /opt/apps/FLINK/flink-current/examples/streaming/TopSpeedWindowing.jar

View client logs

In an EMR cluster environment, the FLINK_LOG_DIR environment variable specifies where Flink client logs are stored. Its default value is /var/log/taihao-apps/flink (the default was /mnt/disk1/log/flink in versions earlier than 3.43.0). If you need to view complete client logs, such as SQL Client logs, you can find the corresponding files in this directory.

Job parameters not taking effect

When you run a Flink job from the command line, place the job parameters after the Flink job JAR package. For example: flink run -d -t yarn-per-job test.jar arg1 arg2.

Resolve "Multiple factories..." error

• Cause

  This error indicates that the classpath contains multiple implementations of a connector. This typically happens when a connector dependency is added to the job's JAR package while you also manually place the same connector dependency in the $FLINK_HOME/lib directory, causing a dependency conflict.

• Solution

  The solution is to remove the duplicate dependency. For detailed troubleshooting steps, see What do I do if a job JAR package conflicts with the cluster's Flink JAR package?

Enable JobManager HA

high-availability: zookeeper
high-availability.zookeeper.quorum: 192.168.**.**:2181,192.168.**.**:2181,192.168.**.**:2181
high-availability.zookeeper.path.root: /flink
high-availability.storageDir: hdfs:///flink/recovery

View Flink job metrics

1. In the EMR console, navigate to the Monitoring page of the target cluster and click Metric Monitoring.

2. From the Dashboard drop-down list, select FLINK.

3. Select the application ID and job ID for the job you want to view. The monitoring metrics for the job will then appear.

   Note

   • The application ID and job ID options are available only if Flink jobs are running in the cluster.

   • Some metrics, such as sourceIdleTime, are generated only if the corresponding source and sink are configured.

Troubleshoot connector issues

For common questions about upstream and downstream storage, see Connectors.

Resolve errors with password-free OSS access

Handle the issue based on the specific error message:

• Error message: java.lang.UnsupportedOperationException: Recoverable writers on Hadoop are only supported for HDFS.

  • Cause: A DataFlow cluster uses the built-in JindoSDK to support password-free access to OSS and APIs like StreamingFileSink. You do not need to perform additional configuration as described in the community documentation. Doing so may cause a dependency conflict that results in this error.

  • Solution: On the job submission machine in your cluster, check the $FLINK_HOME/plugins directory for an oss-fs-hadoop directory. If it exists, delete the directory and resubmit the job.

• Error message: Could not find a file system implementation for scheme 'oss'. The scheme is directly supported by Flink through the following plugin: flink-oss-fs-hadoop. .....

  • Cause: In EMR clusters of version 3.40 and earlier, machines in the master server group other than master-1-1 may be missing Jindo-related JAR packages.

  • Solution:

    • For EMR 3.40 and earlier: Check if Jindo-related JAR packages, such as jindo-flink-4.0.0-full.jar, exist in the $FLINK_HOME/lib directory on the job submission machine. If they are missing, run the following command to copy the required JAR packages to the $FLINK_HOME/lib directory and then resubmit the job.

      cp /opt/apps/extra-jars/flink/jindo-flink-*-full.jar $FLINK_HOME/lib

    • For EMR versions later than 3.40:

      • For Flink on YARN mode: Newer versions have an optimized mechanism for OSS support. Jobs that read from and write to OSS can run normally even if Jindo-related JAR packages are not present in the $FLINK_HOME/lib directory.

      • For other deployment modes: Check if Jindo-related JAR packages, such as jindo-flink-4.0.0-full.jar, exist in the $FLINK_HOME/lib directory of the job submission machine. If they are missing, run the following command to copy them to the $FLINK_HOME/lib directory and then resubmit the job.

        cp /opt/apps/extra-jars/flink/jindo-flink-*-full.jar $FLINK_HOME/lib

Resolve "TaskManager heartbeat timed out" error

• Cause

  The direct cause is a TaskManager heartbeat timeout. You can check the TaskManager logs for specific error messages to pinpoint the exact reason. Other potential causes include insufficient TaskManager heap memory or an out of memory (OOM) error due to a memory leak in the job code. For more information, see How do I resolve the "java.lang.OutOfMemoryError: GC overhead limit exceeded" error?.

• Solution

  If you encounter this error, increase the memory allocation or analyze the job's memory usage to further diagnose the issue.

Resolve "GC overhead limit exceeded" error

• Cause

  This error indicates that the garbage collector (GC) is taking too much time because the memory allocated to the job is insufficient. Common causes include a memory leak in the code (such as a UDF) or the configured memory is insufficient for the job's requirements.

• Solution

  • Before rerunning the job, specify the following JVM parameter using the -D flag to save a heap dump when an OutOfMemoryError occurs: -D env.java.opts="-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof".

  • Add the parameter env.java.opts: -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/dump.hprof to the flink-conf.yaml file to configure heap dumps on OutOfMemoryError.

  After the error occurs again, you can analyze the heap dump file specified by HeapDumpPath using tools like MAT or jvisualvm to determine the root cause.

Zero "Records Received" for single-operator jobs

This is normal. The Flink Records Received metric describes data communication between different operators. When a job is optimized into a single operator, this metric will always be 0.

Enable flame graphs for Flink jobs

A flame graph visualizes the CPU consumption of various methods within a process, helping you identify and resolve performance bottlenecks. Flink has supported flame graphs since version 1.13, but the feature is disabled by default to avoid affecting production jobs. If you need to use flame graphs to analyze job performance, go to the Flink service Configure tab in the EMR console. In the flink-conf.yaml file, add a new configuration item with the parameter rest.flamegraph.enabled and set its value to true. For instructions on adding a configuration item, see Manage configuration items.

For more information about flame graphs, see Flame Graphs.

Resolve "NoSuchFieldError: DEPLOYMENT_MODE" error

• Cause

  Your job's JAR package directly or indirectly includes a flink-core dependency that is incompatible with the Flink version in the cluster, causing a dependency conflict.

• Solution

  Add the following configuration to your pom.xml file to set the scope of the flink-core dependency to provided. This resolves the issue.

  <dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-core</artifactId>
    <!-- Change to your own flink version -->
    <version>1.16.1</version>
    <scope>provided</scope>
  </dependency>

  Note

  You must change the version to your Flink version.

  To further locate the source of this dependency, see What do I do if a job JAR package conflicts with the cluster's Flink JAR package?.