AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Flink之DataSet转换操作（二）

2022-05-18 177
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
实时计算 Flink 版，5000CU*H 3个月
推荐场景：
实时发现最热Github项目
简介： 笔记

（8）Aggregate详解

通过Aggregate Function将一组元素值合并成单个值，可以在整个DataSet数据集上使用。

Java代码实现：

package com.aikfk.flink.dataset.transform;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/7 8:26 下午
 */
public class AggregateJava {
    public static void main(String[] args) throws Exception {
        // 准备环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        DataSet<String> dateSource = env.fromElements(
                "java java spark hive",
                "hive java java spark",
                "java java hadoop"
        );
        /**
         * map
         */
        DataSet<String> mapSource = dateSource.map(new MapFunction<String, String>() {
            @Override
            public String map(String line) throws Exception {
                return line.toUpperCase();
            }
        });
        /**
         * flatmap
         */
        DataSet<Tuple2<String,Integer>> flatmapSource = mapSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
                for (String word : s.split(" ")){
                    collector.collect(new Tuple2<>(word,1));
                }
            }
        });
        /**
         * aggregate
         */
        DataSet<Tuple2<String,Integer>> aggregateSource =  flatmapSource.groupBy(0).aggregate(Aggregations.SUM,1);
        aggregateSource.print();
        /**
         * (HIVE,2)
         * (HADOOP,1)
         * (JAVA,6)
         * (SPARK,2)
         */
    }
}


（9）Join详解

Join的几种方式：


根据指定的条件关联两个数据集，然后根据选择的字段形成一个数据集。关联的key可以通过Key表达式、Key-selector函数、字段位置以及Case Class字段指定。


1.对于两个Tuple类型的数据集可以通过字段位置进行关联，左边数据集的字段通过where方法指定，右边数据集的字段通过equalTo()方式指定

dataSet_1.join(dataSet_2).where(0).equalTo(0)
where(左边的关联的字段).equal(右边关联的字段)
  1. 关联过程中指定自定义Join Funtion
dataSet_1.join(dataSet_2).where(0).equalTo(0){
  (left,right) => (left.id , right.name)
}


3.通过JoinWithTiny或者JoinWithHuge标识数据集的大小

dataSet_1.joinWithTiny(dataSet_2).where(0).equalTo(0)提示F1ink第二个数据集是小数据集
dataSet_1.joinWithHuge(dataSet_2).where(0).equalTo(0)提示Flink第二个数据集是大数据集

Join的优化：

Flink提供了Join算法提示，可以让F1ink更加灵活高效地执行Join操作


1.将第一个数据集广播出去，并转换成HashTable存储，该策略适用于第一个数据集非常小的情况

dataSet_1.join(dataSet_2, JoinHint.BROADCAST_HASH_FIRST).where(0).equalTo(0)
  1. 将第二个数据集广播出去，并转换成HashTable存储，该策略适用于第一个数据集非常小的情况
dataSet_1.join(dataSet_2, JoinHint.BROADCAST_HASH_SECOND).where(0).equalTo(0)
  1. 和不设定Hint相同，将优化工作交给系统自动处理
dataSet_1.join(dataSet_2, JoinHint.OPTIMIZER_CHOOSES).where(0).equalTo(0)

4.将两个数据集重新分区，并将第一个数据集转换成HashTable存储，该策略适用于第一个数据集比第二个数据集小，但两个数据集相对都比较大的情况

dataSet_1.join(dataSet_2, JoinHint.PARTITION_HASH_FIRST).where(0).equalTo(0)
  1. 将两个数据集重新分区，并将第二个数据集转换成HashTable存储，该策略适用于第二个数据集比第一个数据集小，但两个数据集相对都比较大的情况
dataSet_1.join(dataSet_2, JoinHint.PARTITION_HASH_SECOND).where(0).equalTo(0)
  1. 将两个数据集重新分区，并将每个分区排序，该策略适用于两个数据集已经排好序的情况
dataSet_1.join(dataSet_2, JoinHint.PARTITION_SORT_MEGER).where(0).equalTo(0)

join代码实现

数据源：

dept.csv

100,技术部
200,市场部
300,营销部
400,采购部

employee.csv

100,alex,15000
100,jack,34000
200,tony,5000
200,jone,6700
300,cherry,12000
100,lili,8000
400,lucy,7800

Java代码实现：

POJO类

package com.aikfk.flink.base;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/8 3:23 下午
 */
public class EmployeePOJO {
    public String deptId;
    public String name;
    public int salary;
    public EmployeePOJO() {
    }
    public EmployeePOJO(String deptId, String name, int salary) {
        this.deptId = deptId;
        this.name = name;
        this.salary = salary;
    }
    @Override
    public String toString() {
        return "EmployeePOJO{" +
                "deptId='" + deptId + '\'' +
                ", name='" + name + '\'' +
                ", salary=" + salary +
                '}';
    }
}

package com.aikfk.flink.dataset.transform;
import com.aikfk.flink.base.EmployeePOJO;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/8 3:13 下午
 */
public class JoinJava {
    public static void main(String[] args) throws Exception {
        // 准备环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 读取csv数据（方式二：映射成Tuple类，带有两个个字段）
        DataSet<Tuple2<String,String>> deptSource = env.readCsvFile("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/dept.csv")
                .types(String.class,String.class);
        deptSource.print();
        /**
         * (200,市场部)
         * (100,技术部)
         * (400,采购部)
         * (300,营销部)
         */
        // 读取csv数据（方式一：映射POJO类对象）
        DataSet<EmployeePOJO> employeeSource = env.readCsvFile("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee.csv")
                .pojoType(EmployeePOJO.class,"deptId","name","salary");
        employeeSource.print();
        /**
         * EmployeePOJO{deptId='100', name='alex', salary=15000}
         * EmployeePOJO{deptId='200', name='jone', salary=6700}
         * EmployeePOJO{deptId='100', name='lili', salary=8000}
         * EmployeePOJO{deptId='400', name='lucy', salary=7800}
         * EmployeePOJO{deptId='300', name='cherry', salary=12000}
         * EmployeePOJO{deptId='200', name='tony', salary=5000}
         * EmployeePOJO{deptId='100', name='jack', salary=34000}
         */
        /**
         * join() -> map()
         */
        DataSet<Tuple3<String,String,String>> joinResult = deptSource.join(employeeSource).where(0).equalTo("deptId")
                .map(new MapFunction<Tuple2<Tuple2<String, String>, EmployeePOJO>, Tuple3<String, String, String>>() {
                    @Override
                    public Tuple3<String, String, String> map(Tuple2<Tuple2<String, String>, EmployeePOJO> tuple2) throws Exception {
                        return new Tuple3<>(tuple2.f0.f0,tuple2.f0.f1,tuple2.f1.name);
                    }
                });
        joinResult.print();
        /**
         * (100,技术部,jack)
         * (100,技术部,alex)
         * (400,采购部,lucy)
         * (100,技术部,lili)
         * (300,营销部,cherry)
         * (200,市场部,jone)
         * (200,市场部,tony)
         */
    }
}

Scala代码实现：

package com.aikfk.flink.dataset.transform
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
object JoinScala {
  case class employee(deptId:String,name:String,salary:Int)
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment;
    // 读取csv数据（方式二：映射成Tuple类，带有两个个字段）
    val deptSource = env.readCsvFile[(String,String)]("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/dept.csv")
    // 读取csv数据（方式一：映射POJO类对象）
    val employeeSource = env.readCsvFile[employee]("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee.csv")
    // join -> map
    val joinResult = deptSource.join(employeeSource).where(0).equalTo("deptId")
      .map(tuple2 => (tuple2._1._1,tuple2._1._2,tuple2._2.name))
    joinResult.print()
    /**
     * (400,采购部,lucy)
     * (100,技术部,jack)
     * (100,技术部,alex)
     * (100,技术部,lili)
     * (300,营销部,cherry)
     * (200,市场部,tony)
     * (200,市场部,jone)
     */
  }
}


（10）Union详解

合并两个DataSet数据集，两个数据集的数据元素格式必须相同，多个数据集可以连接合并。

dataSet_1.union(dataSet_2)

Union代码实现：

数据源：

employee.csv

100,alex,15000
100,jack,34000
200,tony,5000
200,jone,6700
300,cherry,12000
100,lili,8000
400,lucy,7800

employee2.csv

100,zhang,1400
100,li,3500
200,liu,6000
200,cai,6800
300,wang,13000
100,cao,8900
400,peng,7800

POJO类：

package com.aikfk.flink.base;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/8 3:23 下午
 */
public class EmployeePOJO {
    public String deptId;
    public String name;
    public int salary;
    public EmployeePOJO() {
    }
    public EmployeePOJO(String deptId, String name, int salary) {
        this.deptId = deptId;
        this.name = name;
        this.salary = salary;
    }
    @Override
    public String toString() {
        return "EmployeePOJO{" +
                "deptId='" + deptId + '\'' +
                ", name='" + name + '\'' +
                ", salary=" + salary +
                '}';
    }
}

package com.aikfk.flink.base;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/8 7:40 下午
 */
public class DeptSalaryPOJO {
    public String deptId;
    public int salary;
    public DeptSalaryPOJO() {
    }
    public DeptSalaryPOJO(String deptId, int salary) {
        this.deptId = deptId;
        this.salary = salary;
    }
    @Override
    public String toString() {
        return "DeptSalaryPOJO{" +
                "deptId='" + deptId + '\'' +
                ", salary=" + salary +
                '}';
    }
}

package com.aikfk.flink.base;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/8 7:57 下午
 */
public class DeptPOJO {
    public String deptId;
    public String name;
    public DeptPOJO() {
    }
    public DeptPOJO(String deptId, String name) {
        this.deptId = deptId;
        this.name = name;
    }
    @Override
    public String toString() {
        return "DeptPOJO{" +
                "deptId='" + deptId + '\'' +
                ", name='" + name + '\'' +
                '}';
    }
}

Java代码实现：

package com.aikfk.flink.dataset.transform;
import com.aikfk.flink.base.DeptPOJO;
import com.aikfk.flink.base.DeptSalaryPOJO;
import com.aikfk.flink.base.EmployeePOJO;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
/**
 * @author ：caizhengjie
 * @description：TODO
 * @date ：2021/3/8 7:36 下午
 */
public class UnionJava {
    public static void main(String[] args) throws Exception {
        // 准备环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        // 读取csv数据（方式一：映射POJO类对象）
        DataSet<EmployeePOJO> employeeSource = env.readCsvFile("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee.csv")
                .pojoType(EmployeePOJO.class,"deptId","name","salary");
        // 读取csv数据（方式一：映射POJO类对象）
        DataSet<EmployeePOJO> employee2Source = env.readCsvFile("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee2.csv")
                .pojoType(EmployeePOJO.class,"deptId","name","salary");
        // 读取csv数据（方式一：映射POJO类对象）
        DataSet<DeptPOJO> deptSource = env.readCsvFile("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/dept.csv")
                .pojoType(DeptPOJO.class,"deptId","name");
        /**
         * union() -> map() -> groupBy() -> reduce()
         */
        DataSet<DeptSalaryPOJO> unionResult = employeeSource.union(employee2Source)
                .map(new MapFunction<EmployeePOJO, DeptSalaryPOJO>() {
                    @Override
                    public DeptSalaryPOJO map(EmployeePOJO employeePOJO) throws Exception {
                        return new DeptSalaryPOJO(employeePOJO.deptId,employeePOJO.salary);
                    }
                })
                .groupBy("deptId")
                .reduce(new ReduceFunction<DeptSalaryPOJO>() {
                    @Override
                    public DeptSalaryPOJO reduce(DeptSalaryPOJO t1, DeptSalaryPOJO t2) throws Exception {
                        return new DeptSalaryPOJO(t1.deptId,t1.salary + t2.salary);
                    }
                });
        unionResult.print();
        /**
         * DeptSalaryPOJO{deptId='100', salary=70800}
         * DeptSalaryPOJO{deptId='400', salary=15600}
         * DeptSalaryPOJO{deptId='300', salary=25000}
         * DeptSalaryPOJO{deptId='200', salary=24500}
         */
        /**
         * join() -> map()
         */
        DataSet<Tuple3<String, String, Integer>> joinResult = unionResult.join(deptSource).where("deptId").equalTo("deptId")
            .map(new MapFunction<Tuple2<DeptSalaryPOJO, DeptPOJO>, Tuple3<String, String, Integer>>() {
                @Override
                public Tuple3<String, String, Integer> map(Tuple2<DeptSalaryPOJO, DeptPOJO> tuple2) throws Exception {
                    return new Tuple3<>(tuple2.f0.deptId,tuple2.f1.name,tuple2.f0.salary);
                }
            });
        joinResult.print();
        /**
         * (100,技术部,70800)
         * (400,采购部,15600)
         * (300,营销部,25000)
         * (200,市场部,24500)
         */
    }
}

Scala代码实现：

package com.aikfk.flink.dataset.transform
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
object UnionScala {
  case class employee(deptId:String,name:String,salary:Int)
  def main(args: Array[String]): Unit = {
    val env = ExecutionEnvironment.getExecutionEnvironment;
    // 读取csv数据（方式一：映射POJO类对象）
    val employeeSource = env.readCsvFile[employee]("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee.csv")
    // 读取csv数据（方式一：映射POJO类对象）
    val employeeSource2 = env.readCsvFile[employee]("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/employee2.csv")
    // 读取csv数据（方式二：映射成Tuple类，带有两个个字段）
    val deptSource = env.readCsvFile[(String,String)]("/Users/caizhengjie/IdeaProjects/aikfk_flink/src/main/java/resources/dept.csv")
    /**
     * union() -> map() -> groupBy() -> reduce()
     */
    val unionResult = employeeSource.union(employeeSource2)
      .map(emp => (emp.deptId,emp.salary))
      .groupBy(0)
      .reduce((x ,y) => (x._1,x._2 + y._2))
    unionResult.print()
    /**
     * (100,70800)
     * (400,15600)
     * (300,25000)
     * (200,24500)
     */
    /**
     * join() -> map()
     */
    val joinResult = unionResult.join(deptSource).where(0).equalTo(0)
      .map(tuple => (tuple._1._1,tuple._2._2,tuple._1._2) )
    joinResult.print()
    /**
     * (400,采购部,15600)
     * (100,技术部,70800)
     * (300,营销部,25000)
     * (200,市场部,24500)
     */
  }
}




文章标签：
实时计算 Flink版
Java
流计算
Scala
算法
存储
关键词：
实时计算 Flink版dataset
实时计算 Flink版set
实时计算 Flink版操作
实时计算 Flink版dataset转换操作
相关实践学习
基于Hologres轻松玩转一站式实时仓库
本场景介绍如何利用阿里云MaxCompute、实时计算Flink和交互式分析服务Hologres开发离线、实时数据融合分析的数据大屏应用。
Linux入门到精通
本套课程是从入门开始的Linux学习课程，适合初学者阅读。由浅入深案例丰富，通俗易懂。主要涉及基础的系统操作以及工作中常用的各种服务软件的应用、部署和优化。即使是零基础的学员，只要能够坚持把所有章节都学完，也一定会受益匪浅。
游客wkxim4agoo6le
目录
相关文章
武子康
|
1月前
|
SQL 大数据 API
大数据-118 - Flink DataSet 基本介绍 核心特性 创建、转换、输出等
大数据-118 - Flink DataSet 基本介绍 核心特性 创建、转换、输出等
武子康
55 0
武子康
|
1月前
|
Java Shell 流计算
Flink-02 Flink Java 3分钟上手 Stream SingleOutputStreamOpe ExecutionEnvironment DataSet FlatMapFunction
Flink-02 Flink Java 3分钟上手 Stream SingleOutputStreamOpe ExecutionEnvironment DataSet FlatMapFunction
武子康
24 1
Flink-02 Flink Java 3分钟上手 Stream SingleOutputStreamOpe ExecutionEnvironment DataSet FlatMapFunction
武子康
|
1月前
|
Kubernetes Cloud Native 流计算
Flink-12 Flink Java 3分钟上手 Kubernetes云原生下的Flink集群 Rancher Stateful Set yaml详细 扩容缩容部署 Docker容器编排
Flink-12 Flink Java 3分钟上手 Kubernetes云原生下的Flink集群 Rancher Stateful Set yaml详细 扩容缩容部署 Docker容器编排
武子康
76 3
武子康
|
1月前
|
存储 Java 数据处理
Flink-01 介绍Flink Java 3分钟上手 HelloWorld 和 Stream ExecutionEnvironment DataSet FlatMapFunction
Flink-01 介绍Flink Java 3分钟上手 HelloWorld 和 Stream ExecutionEnvironment DataSet FlatMapFunction
武子康
36 1
三分钟热度的鱼
|
4月前
|
SQL 关系型数据库 MySQL
实时计算 Flink版操作报错合集之从mysql读数据写到hive报错，是什么原因
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
174 5
三分钟热度的鱼
|
4月前
|
SQL 流计算
实时计算 Flink版操作报错合集之怎么向一个未定义列的表中写入数据
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
156 3
三分钟热度的鱼
|
4月前
|
资源调度 分布式计算 Hadoop
实时计算 Flink版操作报错合集之perjob提交给yarn，报错显示无法连接yarn- Connecting to ResourceManager，是什么导致的
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
227 3
三分钟热度的鱼
|
4月前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版操作报错合集之运行mysql to doris pipeline时报错，该如何排查
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
114 2
三分钟热度的鱼
|
4月前
|
Oracle 关系型数据库 Java
实时计算 Flink版操作报错合集之cdc postgres数据库，当表行记录修改后报错，该如何修改
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
123 1
三分钟热度的鱼
|
4月前
|
消息中间件 关系型数据库 MySQL
实时计算 Flink版操作报错合集之整库同步mysql到starRock提交任务异常，该如何处理
在使用实时计算Flink版过程中，可能会遇到各种错误，了解这些错误的原因及解决方法对于高效排错至关重要。针对具体问题，查看Flink的日志是关键，它们通常会提供更详细的错误信息和堆栈跟踪，有助于定位问题。此外，Flink社区文档和官方论坛也是寻求帮助的好去处。以下是一些常见的操作报错及其可能的原因与解决策略。
三分钟热度的鱼
177 1

热门文章

最新文章

  • 1
    Flink on YARN（上）：一张图轻松掌握基础架构与启动流程
  • 2
    Flink三种集群模式，Standalone模式，Flink On YARN，Flink On K8S，这三种模式有啥优缺点，生产环境如何选择呢？
  • 3
    Flink - 新增 BroadcastStream 无 watermark 导致数据流异常
  • 4
    Flink SQL之时态表（Temporal Tables）
  • 5
    Flink 消息聚合处理方案
  • 6
    对于Flink任务的数据丢失问题，你可以从以下几个方面进行排查：
  • 7
    flink 手动维护kafka offset
  • 8
    Flink+Hologres实时数仓在Lazada的建设及应用
  • 9
    10月17日Spark社区直播【Tablestore Spark Streaming Connector -- 海量结构化数据的实时计算和处理】
  • 10
    Flink---13、容错机制（检查点（保存、恢复、算法、配置）、状态一致性、端到端精确一次）
  • 1
    实时计算 Flink版产品使用合集之并行度设置为1时，读取和发送字节数为0，有什么办法可以解决
    61
  • 2
    实时计算 Flink版产品使用合集之是否支持 Oracle 分区表
    52
  • 3
    实时计算 Flink版产品使用合集之ticdc可以控制全量同步的速率吗
    51
  • 4
    实时计算 Flink版产品使用合集之生成OSS URL时长度超过64位的情况如何解决
    60
  • 5
    实时计算 Flink版产品使用合集之在使用Flink SQL向ClickHouse写入数据的过程中出现丢数据或重复数据的情况如何解决
    330
  • 6
    实时计算 Flink版产品使用合集之每个slot对应一个并行度吗
    56
  • 7
    实时计算 Flink版产品使用合集之oracle cdc 抽取新增一张表 可以从savepoint恢复吗
    64
  • 8
    实时计算 Flink版产品使用合集之要将收集到的 MongoDB 数据映射成 JSON 对象而非按字段分割，该怎么操作
    82
  • 9
    实时计算 Flink版产品使用合集之断点续传的步骤是什么
    82
  • 10
    实时计算 Flink版产品使用合集之Flink CDCRangeMap该怎么使用
    57

    • 相关课程

    更多
  • 大数据Flink实时旅游平台环境篇 2020版
  • Apache Flink 入门到实战 - Flink开源社区出品
  • 实时计算 Flink 版产品入门与实操
  • 开源 Flink 极速上手教程
  • 大数据实时计算框架Spark快速入门
  • Apache Flink 入门

    • 相关电子书

    更多
  • Flink CDC Meetup PPT - 龚中强
  • Flink CDC Meetup PPT - 王赫
  • Flink CDC Meetup PPT - 覃立辉

    • 相关实验场景

    更多
  • 基于Hologres+Flink搭建GitHub实时数据大屏
  • 基于Flink+Tair搭建实时监控大屏
  • 5分钟上手Flink MySQL连接器
  • 使用Flink实时发现最热Github项目
    • 下一篇
    无影云桌面