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

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

打包。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怎么将数据写入到文件中”有了更深的了解，不妨来实际操作一番吧！这里是网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！