FlinkX 分片读取原理

女爷i 2023-02-11 13:30 20阅读 0赞

# FlinkX 分片读取原理 #

在数据同步工具中,将数据从源头读取到数据缓存是最重要的一环之一，算是`左膀`。所以在整个流程，从技术场景上，一定要支持数据的分片与并行读取、流控，从业务角度上，需要支持脏值处理与增量读取。

![image-20200523220805741][]

而今天重点来探讨一下分片读取的原理。

### 分片原理 ###

分片是将待读取的数据平均分配，尽量的使各个分片任务均衡，不会让数据倾斜从而导致个别节点的同步压力过大（硬件-网卡、cpu等）。

下面是配置了一个读取通道为3的作业配置示例：

"speed": { 
         "channel": 3,
         "bytes": 0
    },

**重点类&方法**

*  `InputSplit` （输入分片类）表示输入的分片，并且会在运行过程中进行传输，所以需要进行序列化,是Flink的数据读取核心类。
    
    ![image-20200523223305856][]
 *  `BaseRichInputFormat#createInputSplits` 创建分片,会对错误进行捕获，包装输出,
    
    ![image-20200523223659952][]
    
    **此方法实际是FLink中的InputSplitSource**
    
    ![image-20200523223808762][]
    
    org.apache.flink.api.common.io.InputFormat.java
    
    ![image-20200523223909531][]
    
    org.apache.flink.core.io.InputSplitSource.java
    
    由上可以的得知，真实的分片逻辑有具体的实现子类进行提供，将`InputSplit`结果返回给调度系统，而分片的调度由Flink底层进行提供（因为reader读取数据返回的是DataStream）。如下图所示的关系
    
    ![image-20200523224812698][]
    
    Flink、FlinkX在分片逻辑中的关系
 *  `BaseRichInputFormat#createInputSplitsInternal` 创建实际的分片抽象方法，由实际driver创建
    
    ![image-20200523223248125][]

#### 通用JDBC 的分片策略 ####

JDBC分片中的几个概念:

*  splitKey
 *  numPartitions `Math.max(speed.channel,speed.readerChannel)`

具体实现逻辑及代码如下：

@Override
        public InputSplit[] createInputSplitsInternal(int minPart) throws IOException { 
            DistributedJdbcInputSplit[] inputSplits = new DistributedJdbcInputSplit[numPartitions];
    
            if(splitKey != null && splitKey.length()> 0){ 
                Object[][] parmeter = DbUtil.getParameterValues(numPartitions);
                for (int j = 0; j < numPartitions; j++) { 
                    DistributedJdbcInputSplit split = new DistributedJdbcInputSplit(j,numPartitions);
                    List<DataSource> sourceCopy = deepCopyList(sourceList);
                    for (int i = 0; i < sourceCopy.size(); i++) { 
                        sourceCopy.get(i).setSplitByKey(true);
                        sourceCopy.get(i).setParameterValues(parmeter[j]);
                    }
                    split.setSourceList(sourceCopy);
                    inputSplits[j] = split;
                }
            } else { 
                int partNum = sourceList.size() / numPartitions;
                if (partNum == 0){ 
                    for (int i = 0; i < sourceList.size(); i++) { 
                        DistributedJdbcInputSplit split = new DistributedJdbcInputSplit(i,numPartitions);
                        split.setSourceList(Arrays.asList(sourceList.get(i)));
                        inputSplits[i] = split;
                    }
                } else { 
                    for (int j = 0; j < numPartitions; j++) { 
                        DistributedJdbcInputSplit split = new DistributedJdbcInputSplit(j,numPartitions);
                        split.setSourceList(new ArrayList<>(sourceList.subList(j * partNum,(j + 1) * partNum)));
                        inputSplits[j] = split;
                    }
    
                    if (partNum * numPartitions < sourceList.size()){ 
                        int base = partNum * numPartitions;
                        int size = sourceList.size() - base;
                        for (int i = 0; i < size; i++) { 
                            DistributedJdbcInputSplit split = inputSplits[i];
                            split.getSourceList().add(sourceList.get(i + base));
                        }
                    }
                }
            }
    
            return inputSplits;
        }

如上述代码所示：执行逻辑解释如下

![image-20200523232649004][]

**逻辑1：** 如果有splitKey的话，会按照Key进行分片，没有Source 会将数据分为`numPartions`份，每一个channel读取对应的一份数据。

![image-20200523231828682][]

\*\*逻辑2：\*\*将配置的source分为多份。每个channel读取对应的source列表

![image-20200523232413314][]

#### MongoDB的分片策略实现 ####

public InputSplit[] createInputSplitsInternal(int minNumSplits) throws IOException { 
            ArrayList<MongodbInputSplit> splits = new ArrayList<>();
    
            MongoClient client = null;
            try { 
                client = MongodbClientUtil.getClient(mongodbConfig);
                MongoDatabase db = client.getDatabase(mongodbConfig.getDatabase());
                MongoCollection<Document> collection = db.getCollection(mongodbConfig.getCollectionName());
    
                long docNum = filter == null ? collection.countDocuments() : collection.countDocuments(filter);
                if(docNum <= minNumSplits){ 
                    splits.add(new MongodbInputSplit(0,(int)docNum));
                    return splits.toArray(new MongodbInputSplit[splits.size()]);
                }
    
                long size = Math.floorDiv(docNum, minNumSplits);
                for (int i = 0; i < minNumSplits; i++) { 
                    splits.add(new MongodbInputSplit((int)(i * size), (int)size));
                }
    
                if(size * minNumSplits < docNum){ 
                    splits.add(new MongodbInputSplit((int)(size * minNumSplits), (int)(docNum - size * minNumSplits)));
                }
            } finally { 
                MongodbClientUtil.close(client, null);
            }
    
            return splits.toArray(new MongodbInputSplit[splits.size()]);
        }

1.  获取所有当前collection的count,即发出命令如:`{"aggregate": "bond_info", "pipeline": [{"$match": {}}, {"$group": {"_id": 1, "n": {"$sum": 1}}}], "cursor": {}}`
2.  然后按照`minNumSplits`进行分片,如果分片取余不为0，那么会创建`minNumSplits+1`个分片，这里没有考虑多个source 的情况。

### 并行读取 ###

并行读取指的先由分片逻辑分完片返回分片结果后，执行引擎为每一个分片创建一个读取通道，从而达到数据并行到达源的效果。

> 很多人称之为并发读取，个人理解还是称之为并行读取更合适，因为从任务的时间维度上来，属于并行执行。当然可能从数据源的角度上来讲，这里属于并发读取，不对这些概念做些无聊的争执。后续如果有些并行读取或者并发读取指的都是同一个概念。

从上一节的关键类`InputSplitSource`里面可以看出来,其实并行读取底层还是Flink在支持的。所以这里不做过多的探讨，后续Flink关键分析里面见。

[image-20200523220805741]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjA4MDU3NDEucG5n?x-oss-process=image/format,png
[image-20200523223305856]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjMzMDU4NTYucG5n?x-oss-process=image/format,png
[image-20200523223659952]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjM2NTk5NTIucG5n?x-oss-process=image/format,png
[image-20200523223808762]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjM4MDg3NjIucG5n?x-oss-process=image/format,png
[image-20200523223909531]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjM5MDk1MzEucG5n?x-oss-process=image/format,png
[image-20200523224812698]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjQ4MTI2OTgucG5n?x-oss-process=image/format,png
[image-20200523223248125]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMjMyNDgxMjUucG5n?x-oss-process=image/format,png
[image-20200523232649004]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMzI2NDkwMDQucG5n?x-oss-process=image/format,png
[image-20200523231828682]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMzE4Mjg2ODIucG5n?x-oss-process=image/format,png
[image-20200523232413314]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly93dWh1bGFsYS5vc3MtY24tc2hhbmdoYWkuYWxpeXVuY3MuY29tL2Jsb2cvaW1hZ2UtMjAyMDA1MjMyMzI0MTMzMTQucG5n?x-oss-process=image/format,png