flinksql读取kafka写入mysql(flink写数据到数据库)

技术flinksql怎么将数据写入到文件中本篇内容主要讲解“flinksql怎么将数据写入到文件中”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“flinksql怎么将数据写入

本篇内容主要讲解"弗林克索尔怎么将数据写入到文件中",感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习"弗林克索尔怎么将数据写入到文件中"吧!

打包。JD。数据输出;

导入com。JD。数据。传感器读数;

导入组织。阿帕奇。弗林克。API。常见。功能。MapFuncTion

导入组织。阿帕奇。弗林克。流媒体。API。数据流。数据流源;

导入组织。阿帕奇。弗林克。流媒体。API。数据流。SingleOutputStreamOperator

导入组织。阿帕奇。弗林克。流媒体。API。环境。流式执行环境;

导入组织。阿帕奇。弗林克。桌子。API。数据类型;

导入组织。阿帕奇。弗林克。桌子。API。桌子;

导入组织。阿帕奇。弗林克。桌子。API。桥。Java。streamtableenvironment

导入组织。阿帕奇。弗林克。桌子。描述符。文件系统;

导入组织。阿帕奇。弗林克。桌子。描述符。老CSV

导入组织。阿帕奇。弗林克。桌子。描述符。图式;

publicclassFlinkSqlOutputFile{

公共静态void main(String[]args)throwsException {

streamexecutionenvironmentev=streamexecutionenvironment。getexecutionenvironment();

环境。设置并行度(1);

DataStreamSourceStringstream=env。readtextfile('/Users/刘海静/Desktop/flinktestword/AAA。txt’);

//DataStreamSourceStream=env。SocketTextStream(' localhost ',8888);

singletoutstream operator sensorreadingmap=stream。地图(新地图函数字符串,SensorReading(){ 0

publicSensorReadingmap(字符串)throwsException{

String[]split=s.split(',');

returnnewSensorReading(拆分[0],拆分[1],拆分[2]);

}

(=国家统计局标准)国家标准局

p;       });

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//        使用 table api
        Table table = tableEnv.fromDataStream(map);
//        table.printSchema();
        Table select = table.select("a,b");
//        select.printSchema();

//        使用 sql api
//        tableEnv.createTemporaryView("test", map);
//        Table select = tableEnv.sqlQuery(" select a, b from test");
//        select.printSchema();

//        DataStream<SensorReading2> sensorReading2DataStream = tableEnv.toAppendStream(select, SensorReading2.class);
//        sensorReading2DataStream.map(new MapFunction<SensorReading2, Object>() {
//            @Override
//            public Object map(SensorReading2 value) throws Exception {
//                System.out.println(value.a+"   "+ value.b);
//                return null;
//            }
//        });

//        tableEnv.connect(new FileSystem().path("/Users/liuhaijing/IdeaProjects/haijing3/spark/flinksqldemo/output/out.txt"))
//                .withFormat(new Csv())
//                .withSchema(
//                        new Schema()
//                                .field("a", DataTypes.STRING())
//                                .field("b", DataTypes.STRING()))
//                .inAppendMode()
//                .createTemporaryTable("outputTable");
//        select.insertInto("outputTable");

        tableEnv.connect(new FileSystem().path("/Users/liuhaijing/IdeaProjects/haijing3/spark/flinksqldemo/output/out.txt"))
                .withFormat(new OldCsv())
                .withSchema(new Schema()
                                .field("a", DataTypes.STRING())
                ).inAppendMode()
                .createTemporaryTable("outputTable");
        select.insertInto("outputTable");

        env.execute();

    }
}

到此,相信大家对“flinksql怎么将数据写入到文件中”有了更深的了解,不妨来实际操作一番吧!这里是网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!

内容来源网络,如有侵权,联系删除,本文地址:https://www.230890.com/zhan/156282.html

(0)

相关推荐

  • java的getclass方法(java中class使用方法)

    技术Java中的Object.getClass()怎么使用这篇文章主要介绍“Java中的Object.getClass()怎么使用”,在日常操作中,相信很多人在Java中的Object.getClass()怎么使用问题上

    攻略 2021年12月22日
  • https认证是怎样的

    技术https认证是怎样的https认证是怎样的,针对这个问题,这篇文章详细介绍了相对应的分析和解答,希望可以帮助更多想解决这个问题的小伙伴找到更简单易行的方法。使用https过程中有三个角色:CA认证机构: 证书颁发机

    攻略 2021年11月12日
  • 晒被子一般晒多久合适,【被子晒多久】晒被子要晒多久

    技术晒被子一般晒多久合适,【被子晒多久】晒被子要晒多久随着夏季的到来,炽烈的阳光除了让我们感觉到无比炎热之外,还是有一些好处的,那就是方便我们将被子、冬天的厚衣服之类的东西拿出来晾晒杀菌,那么晒被子要晒多久最合适呢?晒被

    生活 2021年10月24日
  • VUE uni-app的常用API是什么

    技术VUE uni-app的常用API是什么VUE uni-app的常用API是什么,相信很多没有经验的人对此束手无策,为此本文总结了问题出现的原因和解决方法,通过这篇文章希望你能解决这个问题。一、路由与页面跳转uni.

    攻略 2021年10月20日
  • k8s service实现原理(k8s调度工作原理)

    技术k8s的原理分析是怎样的这篇文章给大家介绍k8s的原理分析是怎样的,内容非常详细,感兴趣的小伙伴们可以参考借鉴,希望对大家能有所帮助。k8s是一个开源的容器集群管理系统,可以实现容器集群的自动化部署、自动扩缩容、维护

    攻略 2021年12月15日
  • 金箍棒怎么画,孙悟空怎么画,好看的)

    技术金箍棒怎么画,孙悟空怎么画,好看的)一、在画板的中央画出一个倒3字,然后再用一段弧形连接起来画出孙悟空的脸部金箍棒怎么画。二、接下来再画四个小圆圈,按照下图的方法画出孙悟空的眼睛。
    三、接下来我们再画出孙悟空的后脑勺

    生活 2021年10月31日