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

DataFlow clusters include commercial connectors for Hologres, SLS, MaxCompute, DataHub, Elasticsearch, ClickHouse, and more. You can use these alongside open source connectors. The following example shows how to use the 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

DataFlow clusters use the enterprise-grade GeminiStateBackend by default, which delivers 3 to 5 times the performance of open source state backends. 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

The FLINK_LOG_DIR environment variable specifies the Flink client log directory. Its default value is /var/log/taihao-apps/flink (before version 3.43.0, the default was /mnt/disk1/log/flink). To view complete client logs such as SQL Client logs, check 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

  The classpath contains multiple implementations of a connector. This typically happens when a connector dependency is packaged in the job JAR while the same connector also exists 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: DataFlow clusters use the built-in JindoSDK for password-free OSS access and APIs such as StreamingFileSink. No additional configuration from the community documentation is required. Adding it may cause a dependency conflict that triggers 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 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

  A TaskManager heartbeat timeout can result from several causes. Check the TaskManager logs for specific error messages. Common causes include insufficient 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

  The garbage collector (GC) is consuming excessive time because memory is insufficient. Common causes include a memory leak in the code (such as a UDF) or the allocated memory being too low 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

Flame graphs visualize CPU consumption across methods within a process, helping you identify performance bottlenecks. Flink supports 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?.