flink报错踩坑:org.apache.flink.table.catalog.hive.client.HiveShimV100.registerTemporaryFunction

本文涉及的产品
实时计算 Flink 版,5000CU*H 3个月
简介: 当想使用本地开发环境运行flink读写线上hive数据来运行时报错。我使用maven管理的开发环境依赖。由于代码发布到测试环境集群上跑时并没有报错,而测试环境对应的依赖都是使用放在上面的依赖jar的,并不使用本地maven管理的依赖(也就是没有打入项目jar)。所以我猜测是本地运行环境依赖有问题,也就是项目中maven的pom文件的依赖有问题。
出错场景

当想使用本地开发环境运行flink读写线上hive数据来运行时报错。我使用maven管理的开发环境依赖。由于代码发布到测试环境集群上跑时并没有报错,而测试环境对应的依赖都是使用放在上面的依赖jar的,并不使用本地maven管理的依赖(也就是没有打入项目jar)。所以我猜测是本地运行环境依赖有问题,也就是项目中maven的pom文件的依赖有问题。

在多次检查该项目中maven的pom文件导入的依赖和反复查看flink Table API Connector针对hive的官方文档后终于解决了问题。

下面是我处理问题的一些环境版本:

  • flink 1.13.6
  • hive 1.1.0-cdh5.15.1
  • hadoop 2.6.0-cdh5.15.1

这两个问题主要原因是使用maven管理本地环境依赖时依赖没有使用正确。虽说原因很简单,但实际在我们使用排查的时候往往要花费大量的时间解决,而且官方文档的示例主要还是依据主流版本组件给的,所以直接寻找很难找到解决的办法。

错误一:
Exception in thread "main" java.lang.NoSuchMethodError: org.apache.hadoop.hive.common.JavaUtils.closeClassLoadersTo(Ljava/lang/ClassLoader;Ljava/lang/ClassLoader;)Z
at org.apache.flink.table.planner.delegation.hive.HiveParser$HiveParserSessionState.close(HiveParser.java:397)
at org.apache.flink.table.planner.delegation.hive.HiveParser.clearSessionState(HiveParser.java:342)
at org.apache.flink.table.planner.delegation.hive.HiveParser.parse(HiveParser.java:210)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:704)
at com.medbook.foreign.ella.transform.ResOperationLog.transRes(ResOperationLog.java:39)
at com.medbook.foreign.ella.transform.ResOperationLog.main(ResOperationLog.java:45)

这个是在我maven的pom中使用如下依赖时产生的,只列出了重点的依赖,在最下面我会给出完整的flink连接hive读写的依赖

<properties>
    <maven.compiler.source>8</maven.compiler.source>
    <maven.compiler.target>8</maven.compiler.target>
    <flink.version>1.13.6</flink.version>
    <scala.binary.version>2.11</scala.binary.version>
    <hive.version>1.1.0</hive.version>
    <hadoop.version>2.6.0</hadoop.version>
</properties>

<dependencies>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>

    <!--hive-->
    <dependency>
        <groupId>org.apache.hive</groupId>
        <artifactId>hive-exec</artifactId>
        <version>${hive.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>


    <!--hadoop-->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>${hadoop.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>
 
</dependencies>

我们可以看到hive、hadoop的版本时使用原生的版本,而我的集群环境的hive、hadoop版本都使用的时cdh版本所以报错了此种找不到某个方法的错误,这是由于兼容性问题。

    <hive.version>1.1.0</hive.version>
    <hadoop.version>2.6.0</hadoop.version>

随后我把它换成了如下的版本:

<hive.version>1.1.0-cdh5.15.1</hive.version><hadoop.version>2.6.0-cdh5.15.1</hadoop.version>
错误二:
Exception in thread "main" java.lang.NullPointerException
at org.apache.flink.table.catalog.hive.client.HiveShimV100.registerTemporaryFunction(HiveShimV100.java:422)
at org.apache.flink.table.planner.delegation.hive.HiveParser.parse(HiveParser.java:207)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:704)
at com.medbook.foreign.ella.transform.ResOperationLog.transRes(ResOperationLog.java:37)
at com.medbook.foreign.ella.transform.ResOperationLog.main(ResOperationLog.java:43)

在我解决完错误一以后又爆出了错误二,看到错误我了解到可能是flink连接hive依赖的版本有问题。如下这个依赖:

    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-connector-hive_${scala.binary.version}</artifactId>
        <version>${flink.version}</version>
        <!--<scope>provided</scope>-->
    </dependency>

我查阅了官方给的示例maven,如下图。心里想着没错啊,百思不得其解。
在这里插入图片描述

最后突然想到集群环境服务器上我是下载了flink指定hive版本的连接jar包,该不会是maven导入依赖下也需要指定把,于是我再次翻看了文章对应的内容。
在这里插入图片描述

然后我将依赖改为了下面的内容:

<!-- Flink-hive -->
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-sql-connector-hive-1.2.2_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
    <!--<scope>provided</scope>-->
</dependency>

依赖竟然导入成功了,我想这次稳了,果然稳了,数据跑出来了。可以看出我把

flink-connector-hive_${scala.binary.version}

改成了

flink-sql-connector-hive-1.2.2_${scala.binary.version}

就是上面指定的hive版本的连接器jar,所以你们其他版本导入连接依赖时一定也要按照指定版本的来。

下面我的环境下连接hive的最简依赖pom,其他环境也类似,重点是官方示例没有给出的hadoop-client的环境依赖,为啥呢,是因为hive最后也还是要连接hadoop的。相当于我们本地环境需要连接集群的hadoop,所以需要有一个hadoop-client依赖,同时你的本地也配置了连接集群hadoop的hadoop包以及环境变量。而我们官网给出的示例是针对在线上集群环境下的依赖,本身就包含了flink-shaded-hadoop-2-uber-2.7.5-10.0.jar这种连接hadoop的依赖jar,所以就不需要给出啦!但我们本地环境当然需要我们自己配置了。

<?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">
    <parent>
        <artifactId>book-data-warehouse</artifactId>
        <groupId>com.medbook.warehouse</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <flink.version>1.13.6</flink.version>
        <scala.binary.version>2.11</scala.binary.version>
        <hive.version>1.1.0-cdh5.15.1</hive.version>
        <hadoop.version>2.6.0-cdh5.15.1</hadoop.version>
    </properties>

    <dependencies>
    
        <!-- flink-client-->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <!-- Flink-hive -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-sql-connector-hive-1.2.2_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
<!--            <scope>provided</scope>-->
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
<!--            <scope>provided</scope>-->
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_${scala.binary.version}</artifactId>
            <version>${flink.version}</version>
<!--            <scope>provided</scope>-->
        </dependency>

        <!-- Hive Dependency -->
        <dependency>
            <groupId>org.apache.hive</groupId>
            <artifactId>hive-exec</artifactId>
            <version>${hive.version}</version>
<!--            <scope>provided</scope>-->
        </dependency>

        <!--hadoop-->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
<!--            <scope>provided</scope>-->
        </dependency>

    </dependencies>

    <build>

        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>

            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <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></mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>


        </plugins>
    </build>


</project>

最后的提醒:由于大数据各组件纷繁,又存在各种版本以及操作依赖的兼容性,所以看官方给出的文档示例一定要举一反三,找到自己版本对应的依赖包、以及api,才能正确书写、并运行代码。

相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程,适合初学者阅读。由浅入深案例丰富,通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员,只要能够坚持把所有章节都学完,也一定会受益匪浅。
目录
相关文章
|
7天前
|
消息中间件 Java 关系型数据库
实时计算 Flink版操作报错合集之从 PostgreSQL 读取数据并写入 Kafka 时,遇到 "initial slot snapshot too large" 的错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
578 0
|
7天前
|
存储 SQL 关系型数据库
实时计算 Flink版操作报错合集之按时间恢复时,报错:在尝试读取binlog时发现所需的binlog位置不再可用,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
507 0
|
7天前
|
监控 Oracle 关系型数据库
实时计算 Flink版操作报错合集之在配置连接时,添加了scan.startup.mode参数后,出现报错。是什么导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
629 0
|
7天前
|
消息中间件 Oracle 关系型数据库
实时计算 Flink版操作报错合集之连接RabbitMQ时遇到Could not find any factory for identifier 'rabbitmq' that implements 'org.apache.flink.table.factories.DynamicTableFactory'错误,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
260 0
|
7天前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之CDC任务在异常后整个record sent从0初始化开始,是什么导致的
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
375 0
|
7天前
|
Java 关系型数据库 流计算
实时计算 Flink版操作报错合集之配置cats进行从MySQL到StarRocks的数据同步任务时遇到报错,该怎么办
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
292 0
|
SQL 架构师 API
《Apache Flink 知其然,知其所以然》系列视频课程
# 课程简介 目前在我的公众号新推出了《Apache Flink 知其然,知其所以然》的系列视频课程。在内容上会先对Flink整体架构和所适用的场景做一个基础介绍,让你对Flink有一个整体的认识!然后对核心概念进行详细介绍,让你深入了解流计算中一些核心术语的含义,然后对Flink 各个层面的API,如 SQL/Table&DataStreamAPI/PythonAPI 进行详细的介绍,以及
1312 0
《Apache Flink 知其然,知其所以然》系列视频课程
|
12天前
|
监控 大数据 Java
使用Apache Flink进行大数据实时流处理
Apache Flink是开源流处理框架,擅长低延迟、高吞吐量实时数据流处理。本文深入解析Flink的核心概念、架构(包括客户端、作业管理器、任务管理器和数据源/接收器)和事件时间、窗口、状态管理等特性。通过实战代码展示Flink在词频统计中的应用,讨论其实战挑战与优化。Flink作为大数据处理的关键组件,将持续影响实时处理领域。
78 5
|
1月前
|
消息中间件 Java Kafka
实时计算 Flink版操作报错之Apache Flink中的SplitFetcher线程在读取数据时遇到了未预期的情况,该怎么解决
在使用实时计算Flink版过程中,可能会遇到各种错误,了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题,查看Flink的日志是关键,它们通常会提供更详细的错误信息和堆栈跟踪,有助于定位问题。此外,Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
|
25天前
|
数据处理 Apache 流计算

推荐镜像

更多