Apache Beam Programming Guide

1. The following example divides a PCollection into percentile groups.
   
2. 
   
3. // Provide an int value with the desired number of result partitions, and a PartitionFn that represents the partitioning function.
   
4. // In this example, we define the PartitionFn in-line.
   
5. // Returns a PCollectionList containing each of the resulting partitions as individual PCollection objects.
   
6. PCollection<Student> students = ...;
   
7. // Split students up into 10 partitions, by percentile:
   
8. PCollectionList<Student> studentsByPercentile =
   
9.     students.apply(Partition.of(10, new PartitionFn<Student>() {
   
10.         public int partitionFor(Student student, int numPartitions) {
    
11.             return student.getPercentile()  // 0..99
    
12.                  * numPartitions / 100;
    
13.         }}));
    
14. 
    
15. // You can extract each partition from the PCollectionList using the get method, as follows:
    
16. PCollection<Student> fortiethPercentile = studentsByPercentile.get(4);

复制代码

1. Passing side inputs to ParDo:
   
2. 
   
3.   // Pass side inputs to your ParDo transform by invoking .withSideInputs.
   
4.   // Inside your DoFn, access the side input by using the method DoFn.ProcessContext.sideInput.
   
5. 
   
6.   // The input PCollection to ParDo.
   
7.   PCollection<String> words = ...;
   
8. 
   
9.   // A PCollection of word lengths that we'll combine into a single value.
   
10.   PCollection<Integer> wordLengths = ...; // Singleton PCollection
    
11. 
    
12.   // Create a singleton PCollectionView from wordLengths using Combine.globally and View.asSingleton.
    
13.   final PCollectionView<Integer> maxWordLengthCutOffView =
    
14.      wordLengths.apply(Combine.globally(new Max.MaxIntFn()).asSingletonView());
    
15. 
    
16. 
    
17.   // Apply a ParDo that takes maxWordLengthCutOffView as a side input.
    
18.   PCollection<String> wordsBelowCutOff =
    
19.   words.apply(ParDo.withSideInputs(maxWordLengthCutOffView)
    
20.                     .of(new DoFn<String, String>() {
    
21.       public void processElement(ProcessContext c) {
    
22.         String word = c.element();
    
23.         // In our DoFn, access the side input.
    
24.         int lengthCutOff = c.sideInput(maxWordLengthCutOffView);
    
25.         if (word.length() <= lengthCutOff) {
    
26.           c.output(word);
    
27.         }
    
28.   }}));

复制代码

谢谢分享

谢谢分享

Thanks.