Apache Beam Programming Guide

是 否 +2 论坛币

k人 参与回答

经管之家送您一份

应届毕业生专属福利!

求职就业群

赵安豆老师微信：zhaoandou666

经管之家联合CDA

送您一个全额奖学金名额~ !

立即领取

感谢您参与论坛问题回答

经管之家送您两个论坛币！

+2 论坛币

Apache Beam Programming Guide

The Beam Programming Guide is intended for Beam users who want to use the Beam SDKs to create data processing pipelines. It provides guidance for using the Beam SDK classes to build and test your pipeline. It is not intended as an exhaustive reference, but as a language-agnostic, high-level guide to programmatically building your Beam pipeline. As the programming guide is filled out, the text will include code samples in multiple languages to help illustrate how to implement Beam concepts in your pipelines.

Adapt for:

• Java SDK
  
• Python SDK
  

Contents

• Overview
• Creating the Pipeline
• Working with PCollections
  • Creating a PCollection
  • PCollection Characteristics
    • Element Type
    • Immutability
    • Random Access
    • Size and Boundedness
    • Element Timestamps
      
    
    
• Applying Transforms
  • Using ParDo
  • Using GroupByKey
  • Using Combine
  • Using Flatten and Partition
  • General Requirements for Writing User Code for Beam Transforms
  • Side Inputs and Side Outputs
    
• Composite Transforms
• Pipeline I/O
• Running the Pipeline
• Data Encoding and Type Safety
• Working with Windowing
• Working with Triggers
  

扫码加我 拉你入群

请注明：姓名-公司-职位

以便审核进群资格，未注明则拒绝

分享0 收藏0 回帖

关键词：Programming Program apache Guide gram

1. Creating the pipeline
   
2. The Pipeline abstraction encapsulates all the data and steps in your data processing task. Your Beam driver program typically starts by constructing a Pipeline object, and then using that object as the basis for creating the pipeline’s data sets as PCollections and its operations as Transforms.
   
3. 
   
4. To use Beam, your driver program must first create an instance of the Beam SDK class Pipeline (typically in the main() function). When you create your Pipeline, you’ll also need to set some configuration options. You can set your pipeline’s configuration options programatically, but it’s often easier to set the options ahead of time (or read them from the command line) and pass them to the Pipeline object when you create the object.
   
5. 
   
6. The pipeline configuration options determine, among other things, the PipelineRunner that determines where the pipeline gets executed: locally, or using a distributed back-end of your choice. Depending on where your pipeline gets executed and what your specifed Runner requires, the options can also help you specify other aspects of execution.
   
7. 
   
8. To set your pipeline’s configuration options and create the pipeline, create an object of type PipelineOptions and pass it to Pipeline.Create(). The most common way to do this is by parsing arguments from the command-line:
   
9. 
   
10. Java
    
11. Python
    
12. public static void main(String[] args) {
    
13.    // Will parse the arguments passed into the application and construct a PipelineOptions
    
14.    // Note that --help will print registered options, and --help=PipelineOptionsClassName
    
15.    // will print out usage for the specific class.
    
16.    PipelineOptions options =
    
17.        PipelineOptionsFactory.fromArgs(args).create();
    
18. 
    
19.    Pipeline p = Pipeline.create(options);
    
20. The Beam SDKs contain various subclasses of PipelineOptions that correspond to different Runners. For example, DirectPipelineOptions contains options for the Direct (local) pipeline runner, while DataflowPipelineOptions contains options for using the runner for Google Cloud Dataflow. You can also define your own custom PipelineOptions by creating an interface that extends the Beam SDKs’ PipelineOptions class.

复制代码

1. Reading from an external source
   
2. 
   
3. To read from an external source, you use one of the Beam-provided I/O adapters. The adapters vary in their exact usage, but all of them from some external data source and return a PCollection whose elements represent the data records in that source.
   
4. 
   
5. Each data source adapter has a Read transform; to read, you must apply that transform to the Pipeline object itself. TextIO.Read, for example, reads from an external text file and returns a PCollection whose elements are of type String, each String represents one line from the text file. Here’s how you would apply TextIO.Read to your Pipeline to create a PCollection:
   
6. 
   
7. Java
   
8. Python
   
9. public static void main(String[] args) {
   
10.     // Create the pipeline.
    
11.     PipelineOptions options =
    
12.         PipelineOptionsFactory.fromArgs(args).create();
    
13.     Pipeline p = Pipeline.create(options);
    
14. 
    
15.     PCollection<String> lines = p.apply(
    
16.       "ReadMyFile", TextIO.Read.from("protocol://path/to/some/inputData.txt"));
    
17. }

复制代码

1. Creating a PCollection from in-memory data
   
2. 
   
3. To create a PCollection from an in-memory Java Collection, you use the Beam-provided Create transform. Much like a data adapter’s Read, you apply Create directly to your Pipeline object itself.
   
4. 
   
5. As parameters, Create accepts the Java Collection and a Coder object. The Coder specifies how the elements in the Collection should be encoded.
   
6. 
   
7. The following example code shows how to create a PCollection from an in-memory List:
   
8. 
   
9. public static void main(String[] args) {
   
10.     // Create a Java Collection, in this case a List of Strings.
    
11.     static final List<String> LINES = Arrays.asList(
    
12.       "To be, or not to be: that is the question: ",
    
13.       "Whether 'tis nobler in the mind to suffer ",
    
14.       "The slings and arrows of outrageous fortune, ",
    
15.       "Or to take arms against a sea of troubles, ");
    
16. 
    
17.     // Create the pipeline.
    
18.     PipelineOptions options =
    
19.         PipelineOptionsFactory.fromArgs(args).create();
    
20.     Pipeline p = Pipeline.create(options);
    
21. 
    
22.     // Apply Create, passing the list and the coder, to create the PCollection.
    
23.     p.apply(Create.of(LINES)).setCoder(StringUtf8Coder.of())
    
24. }

复制代码

1. Applying ParDo
   
2. 
   
3. Like all Beam transforms, you apply ParDo by calling the apply method on the input PCollection and passing ParDo as an argument, as shown in the following example code:
   
4. 
   
5. // The input PCollection of Strings.
   
6. PCollection<String> words = ...;
   
7. 
   
8. // The DoFn to perform on each element in the input PCollection.
   
9. static class ComputeWordLengthFn extends DoFn<String, Integer> { ... }
   
10. 
    
11. // Apply a ParDo to the PCollection "words" to compute lengths for each word.
    
12. PCollection<Integer> wordLengths = words.apply(
    
13.     ParDo
    
14.     .of(new ComputeWordLengthFn()));        // The DoFn to perform on each element, which
    
15.                                             // we define above.

复制代码

1. Creating a DoFn
   
2. 
   
3. The DoFn object that you pass to ParDo contains the processing logic that gets applied to the elements in the input collection. When you use Beam, often the most important pieces of code you’ll write are these DoFns–they’re what define your pipeline’s exact data processing tasks.
   
4. 
   
5. Note: When you create your DoFn, be mindful of the General Requirements for Writing User Code for Beam Transforms and ensure that your code follows them.
   
6. A DoFn processes one element at a time from the input PCollection. When you create a subclass of DoFn, you’ll need to provide type paraemters that match the types of the input and output elements. If your DoFn processes incoming String elements and produces Integer elements for the output collection (like our previous example, ComputeWordLengthFn), your class declaration would look like this:
   
7. 
   
8. static class ComputeWordLengthFn extends DoFn<String, Integer> { ... }

复制代码

1. Using Combine
   
2. 
   
3. Combine is a Beam transform for combining collections of elements or values in your data. Combine has variants that work on entire PCollections, and some that combine the values for each key in PCollections of key/value pairs.
   
4. 
   
5. When you apply a Combine transform, you must provide the function that contains the logic for combining the elements or values. The combining function should be commutative and associative, as the function is not necessarily invoked exactly once on all values with a given key. Because the input data (including the value collection) may be distributed across multiple workers, the combining function might be called multiple times to perform partial combining on subsets of the value collection. The Beam SDK also provides some pre-built combine functions for common numeric combination operations such as sum, min, and max.
   
6. 
   
7. Simple combine operations, such as sums, can usually be implemented as a simple function. More complex combination operations might require you to create a subclass of CombineFn that has an accumulation type distinct from the input/output type.
   
8. 
   
9. Simple combinations using simple functions
   
10. 
    
11. The following example code shows a simple combine function.
    
12. 
    
13. // Sum a collection of Integer values. The function SumInts implements the interface SerializableFunction.
    
14. public static class SumInts implements SerializableFunction<Iterable<Integer>, Integer> {
    
15.   @Override
    
16.   public Integer apply(Iterable<Integer> input) {
    
17.     int sum = 0;
    
18.     for (int item : input) {
    
19.       sum += item;
    
20.     }
    
21.     return sum;
    
22.   }
    
23. }

复制代码

1. The following example code shows how to define a CombineFn that computes a mean average:
   
2. 
   
3. public class AverageFn extends CombineFn<Integer, AverageFn.Accum, Double> {
   
4.   public static class Accum {
   
5.     int sum = 0;
   
6.     int count = 0;
   
7.   }
   
8. 
   
9.   @Override
   
10.   public Accum createAccumulator() { return new Accum(); }
    
11. 
    
12.   @Override
    
13.   public Accum addInput(Accum accum, Integer input) {
    
14.       accum.sum += input;
    
15.       accum.count++;
    
16.       return accum;
    
17.   }
    
18. 
    
19.   @Override
    
20.   public Accum mergeAccumulators(Iterable<Accum> accums) {
    
21.     Accum merged = createAccumulator();
    
22.     for (Accum accum : accums) {
    
23.       merged.sum += accum.sum;
    
24.       merged.count += accum.count;
    
25.     }
    
26.     return merged;
    
27.   }
    
28. 
    
29.   @Override
    
30.   public Double extractOutput(Accum accum) {
    
31.     return ((double) accum.sum) / accum.count;
    
32.   }
    
33. }

复制代码

1. Combining a PCollection into a single value
   
2. 
   
3. Use the global combine to transform all of the elements in a given PCollection into a single value, represented in your pipeline as a new PCollection containing one element. The following example code shows how to apply the Beam provided sum combine function to produce a single sum value for a PCollection of integers.
   
4. 
   
5. // Sum.SumIntegerFn() combines the elements in the input PCollection.
   
6. // The resulting PCollection, called sum, contains one value: the sum of all the elements in the input PCollection.
   
7. PCollection<Integer> pc = ...;
   
8. PCollection<Integer> sum = pc.apply(
   
9.    Combine.globally(new Sum.SumIntegerFn()));

复制代码

1. The following example shows how to apply a Flatten transform to merge multiple PCollection objects.
   
2. 
   
3. // Flatten takes a PCollectionList of PCollection objects of a given type.
   
4. // Returns a single PCollection that contains all of the elements in the PCollection objects in that list.
   
5. PCollection<String> pc1 = ...;
   
6. PCollection<String> pc2 = ...;
   
7. PCollection<String> pc3 = ...;
   
8. PCollectionList<String> collections = PCollectionList.of(pc1).and(pc2).and(pc3);
   
9. 
   
10. PCollection<String> merged = collections.apply(Flatten.<String>pCollections());

复制代码