ApacheBeam管道Java:记录未按顺序写入目标文件。
创始人
2024-09-05 11:30:06
0

问题源于具有并行化和异步处理功能的Apache Beam框架可能无法保证数据流在目标文件中以正确的顺序写入。为了解决这个问题,可以使用有序写入器(OrderedWriter)来确保记录按正确的顺序写入目标文件。以下是一个Java代码示例,演示如何使用有序写入器来解决此问题:

PCollection records = ...; // input PCollection
PCollectionView filenamesView = ...; // PCollectionView of the filenames to write to
final TupleTag doneTag = new TupleTag<>();
//Create a new PCollection by assigning a unique, increasing ID to each element
PCollection> keyedRecords =
        records.apply("AddUniqueIds", WithKeys.of((Void) null)).setCoder(KvCoder.of(VarIntCoder.of(), MyRecordCoder.of()));
//Group all elements assigned with the same key (null), and sort all records by id.
PCollection sortedRecords =
        PCollectionList.of(keyedRecords.apply(GroupByKey.create()))
                .apply(ParDo.of(new DoFn>, MyRecord>() {
                    @ProcessElement
                    public void processElement(ProcessContext context) {
                        List sorted = new ArrayList<>();
                        for (MyRecord r : context.element().getValue()) {
                            sorted.add(r);
                        }
                        Collections.sort(sorted, new Comparator() {
                            @Override
                            public int compare(MyRecord o1, MyRecord o2) {
                                // Assumes that MyRecord has a method that returns its id as an int.
                                return Long.compare(o1.getId(), o2.getId());
                            }
                        });
                        for (MyRecord r : sorted) {
                            context.output(r);
                        }
                    }
                })).setCoder(MyRecordCoder.of());
// Create a new representation of the input PCollection where each element is a tuple containing
// the filename and the record it belongs to.
// For example, if the input file was records 0, 1, 2, 3, 4, 5, 6, the output would be:
// ("file0", record 0), ("file1", record 1), ("file2", record 2), ("file3", record 3), ...
final PCollection> keyedOutput = sortedRecords
        .apply("AssignFilename", ParDo.of(new DoFn>() {
            @ProcessElement
            public void processElement(ProcessContext context) {
                List filenames = context.sideInput

相关内容

热门资讯

一次性透视!aapoker破解... 一次性透视!aapoker破解侠是真的“推荐开挂辅助器”1、aapoker破解侠是真的系统规律教程、...
第一次性发现!指尖四川辅助脚本... 第一次性发现!指尖四川辅助脚本“解谜开挂辅助插件”亲,关键说明,指尖四川辅助脚本赛季回归,指尖四川辅...
六次性领会!指尖四川辅助脚本视... 六次性领会!指尖四川辅助脚本视频“揭幕开挂辅助教程”1、上手简单,内置详细流程视频教学,新手小白可以...
9次性普及!掌电竞技辅助器“解... 9次性普及!掌电竞技辅助器“解密开挂辅助插件”一、掌电竞技辅助器AI软件牌型概率发牌机制”必胜“技巧...
第8次性掌握!赣牌圈修改器“推... 第8次性掌握!赣牌圈修改器“推荐开挂辅助app”1、上手简单,内置详细流程视频教学,新手小白可以快速...
三次性了解!大菠萝789辅助“... 三次性了解!大菠萝789辅助“解迷开挂辅助神器”1、首先打开大菠萝789辅助最新版本,在大菠萝789...
第二次性私人局!欢聚水鱼神器“... 第二次性私人局!欢聚水鱼神器“曝光开挂辅助器”1、超多福利:超高返利,海量正版游戏,欢聚水鱼神器系统...
3次性理解!闲逸辅助软件“专业... 3次性理解!闲逸辅助软件“专业开挂辅助插件”闲逸辅助软件辅助器中分为三种模型:闲逸辅助软件软件透明挂...
第五次性掌握!皮皮游戏辅助工具... 第五次性掌握!皮皮游戏辅助工具“揭露开挂辅助教程”在进入皮皮游戏辅助工具辅助挂后,参与本局比赛的八名...
1次性晓得!沧海十三水私人局辅... 1次性晓得!沧海十三水私人局辅助器“分享开挂辅助器”沧海十三水私人局辅助器是一种具有地方特色的麻将游...