微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦！

flink消费kafka数据写入阿里hologres(java版)

分类：常见问题作者：moshang_3377

一、需求背景概述

由于公司业务需要将kafka的数据写入到hologres中，但是发现flinksql那种简便的方式会各种报错(很有可能是我包引用的不对，但是找了阿里支持也没有解决)，无奈只能使用flink-core进行编写。当然，代码并没有那么完美，方法也不一定是最优，有小伙伴有更好的方法和实践经验的话欢迎交流。

二、pom依赖

由于公司使用的版本是flink1.11且项目中还有别的代码，所以pom文件中会有多余的依赖，大家可以根据实际情况各取所需，至少这份代码使用下面的依赖没有发生冲突或者错误。

<?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>******</groupId>
    <artifactId>******</artifactId>
    <version>1.0-SNAPSHOT</version>


    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <scala.binary.version>2.11.8</scala.binary.version>
        <jdk.compile.version>1.8</jdk.compile.version>
        <flink.version>1.11.3</flink.version>
        <maven.version>3.3.3</maven.version>
        <hadoop.version>2.8.4</hadoop.version>
        <hbase.version>2.2.0</hbase.version>
        <hologres.client.version>2.1.1</hologres.client.version>
    </properties>

    <dependencies>

        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.8</version>
        </dependency>

        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-compiler</artifactId>
            <version>2.11.8</version>
        </dependency>
        <!-- flink start -->
        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-clients_2.11</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-streaming-java_2.11</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-core</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-scala_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-java</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-scala_2.11</artifactId>
            <version>${flink.version}</version>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.flink</groupId>
                    <artifactId>flink-streaming-java_2.11</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-streaming-java_2.11</artifactId>
            <version>1.11.3</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-kafka-0.10_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-statebackend-rocksdb_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-planner-blink_2.11</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-scala-bridge_2.11</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-api-java-bridge_2.11</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-json</artifactId>
            <version>${flink.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-connector-jdbc_2.11</artifactId>
            <version>${flink.version}</version>
        </dependency>
        
		<dependency>
            <groupId>com.alibaba.hologres</groupId>
            <artifactId>hologres-connector-flink-1.11</artifactId>
            <version>1.0.0</version>
        </dependency>

        <dependency>
            <groupId>org.apache.flink</groupId>
            <artifactId>flink-table-common</artifactId>
            <version>${flink.version}</version>
            <scope>provided</scope>
        </dependency>
        
        <!-- flink end -->

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>fastjson</artifactId>
            <version>1.2.73</version>
        </dependency>

        <dependency>
            <groupId>com.typesafe</groupId>
            <artifactId>config</artifactId>
            <version>1.3.3</version>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.1.6</version>
        </dependency>

        <dependency>
            <groupId>org.apache.httpcomponents</groupId>
            <artifactId>httpclient</artifactId>
            <version>4.5.2</version>
        </dependency>

        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
            <version>3.4.5</version>
        </dependency>

        <dependency>
            <groupId>MysqL</groupId>
            <artifactId>MysqL-connector-java</artifactId>
            <version>8.0.30</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.25</version>
            <scope>compile</scope>
        </dependency>

        <dependency>
            <groupId>com.alibaba</groupId>
            <artifactId>druid</artifactId>
            <version>1.1.6</version>
        </dependency>

        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-simple</artifactId>
            <version>1.7.25</version>
            <scope>compile</scope>
        </dependency>

        <!-- https://mvnrepository.com/artifact/com.google.code.gson/gson -->
        <dependency>
            <groupId>com.google.code.gson</groupId>
            <artifactId>gson</artifactId>
            <version>2.8.5</version>
        </dependency>

    </dependencies>
    <build>
        <plugins>
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                        <configuration>
                            <includes>
                                <include>**/*.scala</include>
                                <include>**/*.java</include>
                            </includes>
                        </configuration>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.1.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <artifactSet>
                                <excludes>
                                    <exclude>com.google.code.findbugs:jsr305</exclude>
                                    <exclude>org.slf4j:*</exclude>
                                    <exclude>log4j:*</exclude>
                                </excludes>
                            </artifactSet>
                            <filters>
                                <filter>
                                    <!-- Do not copy the signatures in the Meta-INF folder.
                                    Otherwise, this might cause SecurityExceptions when using the JAR. -->
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>Meta-INF/*.SF</exclude>
                                        <exclude>meta-inf/*.DSA</exclude>
                                        <exclude>meta-inf/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

三、代码实现

为了方便我把所有代码放到了一起，这样也方便查找，当然了，关键的信息我会隐去，大家可以根据自己的情况替换成自己的配置。(在代码中其实需要将kafka的数据进行转换然后写到holo)

代码依赖

在看别的博客的时候有时候发现有的没有依赖包，尤其在比较着急看成效的时候还挺麻烦的，所以为了方便自己和大家，我把import信息也加上。

import com.alibaba.fastjson.JSON;
import com.alibaba.ververica.connectors.hologres.api.HologresTableSchema;
import com.alibaba.ververica.connectors.hologres.config.HologresConnectionParam;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.Typeinformation;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.filesystem.FsstateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.alibaba.hologres.client.model.Record;
import com.alibaba.ververica.connectors.hologres.api.HologresRecordConverter;
import com.alibaba.ververica.connectors.hologres.config.HologresConfigs;
import com.alibaba.ververica.connectors.hologres.jdbc.HologresJDBCWriter;
import com.alibaba.ververica.connectors.hologres.sink.HologresOutputFormat;
import org.apache.flink.streaming.api.functions.sink.OutputFormatSinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer010;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.TableSchema;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.CharBuffer;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.text.SimpleDateFormat;
import java.util.Map;
import java.util.Properties;

正式代码

public class sensorLog2holo {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setStateBackend(new FsstateBackend("checkpoint地址"));
        env.enableCheckpointing(5000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //kafka配置文件
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "******");
        properties.setProperty("group.id", "******");
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        properties.setProperty("auto.offset.reset", "******");

        DataStreamSource<SensorLog> kafkaSource =
                env.addSource(new FlinkKafkaConsumer010<SensorLog>("topic name", new SensorLogSerializer(), properties));


        //holo配置
        Configuration configuration = new Configuration();
        configuration.set(HologresConfigs.ENDPOINT, "******");
        configuration.set(HologresConfigs.DATABASE, "******");
        configuration.set(HologresConfigs.USERNAME, "******");
        configuration.set(HologresConfigs.PASSWORD, ""******");");
        configuration.set(HologresConfigs.TABLE, "schema名称.表名");
        HologresConnectionParam hologresConnectionParam = new HologresConnectionParam(configuration);

        /*
        *目标表table schema，下面是示例
        */
        TableSchema schema =
                TableSchema.builder()
                        .field("event", DataTypes.STRING())
                        .field("distinct_id", DataTypes.STRING())
                        .field("appname", DataTypes.STRING())
                        .build();

        kafkaSource.addSink(
                new OutputFormatSinkFunction<SensorLog>(
                        new HologresOutputFormat<>(
                                hologresConnectionParam,
                                new HologresJDBCWriter<>(
                                        hologresConnectionParam,
                                        schema,
                                        new RecordConverter(hologresConnectionParam)))));

       env.execute("sensorLog2holo");

    }


    /**
     * 将用户POJO数据转换至Hologres Record的实现.
     */
    public static class RecordConverter implements HologresRecordConverter<SensorLog, Record> {
        private HologresConnectionParam hologresConnectionParam;
        private HologresTableSchema tableSchema;

        public RecordConverter(HologresConnectionParam hologresConnectionParam) {
            this.hologresConnectionParam = hologresConnectionParam;
        }

        @Override
        public Record convertFrom(SensorLog record) {
            if (tableSchema == null) {
                this.tableSchema =
                        HologresTableSchema.get(hologresConnectionParam.getJdbcoptions());
            }
            Record result = new Record(tableSchema.get());
            result.setobject(0, record.getEvent());
            result.setobject(1, record.getdistinct_id());
            result.setobject(3, record.getProperties().get("appname"));
            return result;
        }

        @Override
        public SensorLog convertTo(Record record) {
            throw new UnsupportedOperationException("No need to implement");
        }

        @Override
        public Record convertToPrimaryKey(SensorLog record) {
            throw new UnsupportedOperationException("No need to implement");
        }
    }

    /**
     * 自定义序列化
     */
    static class SensorLogSerializer implements DeserializationSchema<SensorLog> {
        private static final long serialVersionUID = 1L;

        @Override
        public SensorLog deserialize(byte[] message) throws IOException {
            ByteBuffer buffer = ByteBuffer.wrap(message).order(ByteOrder.LITTLE_ENDIAN);

            String mess = this.byteBuffertoString(buffer);
            //封装为POJO类
            SensorLog sensorLog = JSON.parSEObject(mess, SensorLog.class);
            return sensorLog;
        }

        @Override
        public boolean isEndOfStream(SensorLog nextElement) {
            return false;
        }

        @Override
        public Typeinformation<SensorLog> getProducedType() {
            return Typeinformation.of(SensorLog.class);
        }

        /**
         * 将ByteBuffer类型转换为String类型
         *
         * @param buffer
         * @return
         */
        static public String byteBuffertoString(ByteBuffer buffer) {
            Charset charset = null;
            CharsetDecoder decoder = null;
            CharBuffer charBuffer = null;
            try {
                charset = Charset.forName("UTF-8");
                decoder = charset.newDecoder();
                charBuffer = decoder.decode(buffer.asReadOnlyBuffer());
                return charBuffer.toString();
            } catch (Exception ex) {
                ex.printstacktrace();
                return "";
            }
        }
    }


    /*
    *kafka数据对应的实体类，需要有getter&setter
    */
    static class SensorLog {
        *****
    }
}

原文地址：https://www.jb51.cc/wenti/3281922.html

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 dio@foxmail.com 举报，一经查实，本站将立刻删除。

javajavakafka

相关推荐

使用Nvdia显卡驱动自带的录屏功能录制游戏
显卡天梯图2024最新版 显卡性能天梯图2024最新排行榜
显卡天梯图2024最新版，显卡是电脑进行图形处理的重要设备，显卡的性能高低决定了电脑的性能高低。显卡天梯图用数据直观地为我们展示了具体性能数据，让我们可以根据这些参数进行显卡性能的选择，让我们可以获得符合自己使用需求的显卡
初始化电脑时出现问题怎么办 初始化电脑时出现问题解决方法
初始化电脑时出现问题怎么办，可以使用win系统的安装介质，连接电脑后重启，选择从介质中引导，然后点击修复计算机-疑难解答-高级选项-自动修复，等待一段时间就可以了。有些小伙伴在初始化电脑是碰到了问题，下面小编给大家带来了初始化电脑时出现问题解决方法。
todesk远程开机怎么设置 todesk远程开机教程
todesk远程开机怎么设置，两台电脑要在同一局域网内，然后需要修改BIOS，还要修改电源选项。我们在使用todesk软件时，有些小伙伴想要进行远程开机，那应该怎么设置呢，下面小编就给大家带来了todesk远程开机教程，一起来看看吧。
油猴谷歌插件怎么安装 油猴谷歌插件安装教程
油猴谷歌插件怎么安装，可以通过谷歌应用商店进行安装，需要才能打开应用商店，或者是下载插件到本地，解压后把Tampermonkey.crx文件拖到谷歌浏览器的程序扩展页面进行安装。小伙伴想知道谷歌浏览器怎么安装油猴，下面小编给大家带来了油猴谷歌插件安装教程。
16g虚拟内存设置多少合适 16g虚拟内存最佳设置方法
虚拟内存这个名词想必很多人都听说过，我们在使用电脑的时候，特别是玩大型游戏时候，如果内存空间不足则会开启使用虚拟内存，这样就能够防止死机的情况。有不少的新用户在安装系统的时候，不知道16G的内存设置多大的虚拟内存，下面小编就详细给大家讲解下多少才
win11本地账户怎么改名 win11本地账户改名图文介绍
win11本地账户怎么改名?win11很多操作都变了样，用户如果想要修改用户名称可能找不到具体的位置在哪。win11系统本地账户的名称都是默认的，很多用户为了有特征性想要更改一些有趣的名称，这样看起来更有个性点。win11本地账户怎么改名，我们可以通过打开控制面板，然后
网速怎么看快慢k/s,b/s是什么意思 网速怎么看快慢k/s,b/s哪个快
网速怎么看快慢k s,b s是什么意思，k b就是网速1kb每秒，b s就是网速1b每秒,1kb=1024b。很多小伙伴在下载或上传文件时都会看着kb s，b s等等单位，这些单位是什么意思呢，哪个比较快呢，下面小编给大家带来了网速怎么看快慢k s,b s哪个快的相关资讯。
笔记本声卡驱动怎么更新 笔记本声卡驱动更新方法
我们在使用电脑的时候，经常需要对各种驱动进行更新，如果长期不更新会导致无法正常的使用。有不少的用户在问，该怎么更新声卡驱动？其实很简单，我们右键“我的电脑”打开“设备管理器”，找到“声音”选项并右键“realtek audio”更新驱动程序即可，具体如下。
应用程序无法正常启动0x0000142怎么办 0x0000142错误代码解决办法
应用程序无法正常启动0x0000142怎么办?我们的电脑在日经月累的使用下，难免会出现各种问题，最近有小伙伴反应自己的电脑出现应用程序无法正常启动0x0000142的错误代码，这要怎么解决呢?下面小编就告诉大家电脑遇到0x0000142错误代码的解决办法。
苹果市值2025年有望达4万亿美元
pythonJavaScriptjavaHTMLPHPreactjsC#AndroidCSSNode.jssqlrpython-3.xMysqLjQueryc++pandasFlutterangularIOSdjangolinuxswifttypescript路由器JSON路由器设置无线路由器h3c华三华三路由器设置华三路由器电脑软件教程arraysdocker软件图文教程Cvue.jslaravelspring-boot