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

相关内容

热门资讯

必备攻略透视器脚本!追忆多功能... 必备攻略透视器脚本!追忆多功能透视器脚本,追忆果然是有透视器脚本1、游戏颠覆性的策略玩法,独创攻略技...
专业讨论透视挂方法!天津麻将多... 专业讨论透视挂方法!天津麻将多功能透视挂方法,天津麻将果然有透视挂方法1、下载好天津麻将正确养号方法...
一分钟秒懂辅助器方法!王钓麻将... 一分钟秒懂辅助器方法!王钓麻将多功能辅助器方法,王钓麻将总是是真的有辅助器方法王钓麻将脚本下载中分为...
9分钟了解开挂方法!良人麻将多... 9分钟了解开挂方法!良人麻将多功能开挂方法,良人麻将切实存在有开挂方法1、良人麻将免费脚本咨询教程、...
一分钟了解透视器软件!博雅地方... 一分钟了解透视器软件!博雅地方棋牌多功能透视器软件,博雅地方棋牌都是存在有透视器软件1、下载好博雅地...
如何分辨真伪透视器app!广东... 如何分辨真伪透视器app!广东麻将多功能透视器app,广东麻将果然有透视器app1、广东麻将有没有辅...
分享一款辅助脚本!小巷娱乐多功... 分享一款辅助脚本!小巷娱乐多功能辅助脚本,小巷娱乐原来有辅助脚本小薇(辅助器软件下载)致您一封信;亲...
玩家必看科普辅助器攻略!青橙竞... 玩家必看科普辅助器攻略!青橙竞技多功能辅助器攻略,青橙竞技果然存在有辅助器攻略1)青橙竞技免费钻石:...
详细说明辅助器方法!打两圈兴化... 详细说明辅助器方法!打两圈兴化麻将多功能辅助器方法,打两圈兴化麻将一贯是真的有辅助器方法运打两圈兴化...
必备攻略开挂工具!52麻将多功... 必备攻略开挂工具!52麻将多功能开挂工具,52麻将一直有开挂工具1、超多福利:超高返利,海量正版游戏...