Apache Beam 大数据处理框架

pip install apache-beam

<dependency>  <groupId>org.apache.beam</groupId>  <artifactId>beam-sdks-java-core</artifactId>  <version>2.50.0</version></dependency>

go get github.com/apache/beam/sdks/v2/go/pkg/beam

docker build -t beam-build -f dev-support/docker/Dockerfile dev-support/docker

import apache_beam as beam
with beam.Pipeline() as p:  (p | beam.Create(['Hello Beam', 'Welcome to Beam'])     | beam.FlatMap(lambda line: line.split())     | beam.combiners.Count.PerElement()     | beam.Map(print))

Pipeline p = Pipeline.create();p.apply(KafkaIO.read()    .withBootstrapServers("kafka:9092")    .withTopic("input-topic")) .apply(Window.into(FixedWindows.of(Duration.standardMinutes(1)))) .apply(Count.perElement()) .apply(KafkaIO.write()    .withBootstrapServers("kafka:9092")    .withTopic("output-topic"));p.run();

python wordcount.py \  --runner DataflowRunner \  --project your-project \  --region us-central1

class ParDo(beam.PTransform):  def __init__(self, fn, *args, **kwargs):    self.fn = fn    self.args = args    self.kwargs = kwargs
  def expand(self, pcoll):    return pcoll | beam.Map(self.fn, *self.args, **self.kwargs)
class GroupByKey(beam.PTransform):  def expand(self, pcoll):    return pcoll | beam.CombinePerKey(beam.combiners.ToListCombineFn())

public void run() {  PipelineOptions options = getOptions();  PipelineRunner runner = PipelineRunner.fromOptions(options);  runner.run(this);    // 优化执行计划  Pipeline p = optimizePipeline(this);  // 分阶段执行  List<StageResult> results = executeStages(p);  // 监控执行状态  monitorExecution(results);}

type WindowFn interface {  AssignWindows(ts typex.EventTime) []Window  MergeWindows(windows []Window) []Window}
type FixedWindows struct {  Size time.Duration}
func (w *FixedWindows) AssignWindows(ts typex.EventTime) []Window {  start := ts.Truncate(w.Size)  return []Window{NewIntervalWindow(start, start.Add(w.Size))}}