分布式数据处理框架：Apache Beam_org.apache.beam.sdk.pipeline

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
	xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
	<modelVersion>4.0.0</modelVersion>
 
	<groupId>william-lib</groupId>
	<artifactId>wordcount</artifactId>
	<version>0.0.1-SNAPSHOT</version>
	<packaging>jar</packaging>
 
	<name>wordcount</name>
	<url>http://maven.apache.org</url>
 
	<properties>
		<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
	</properties>
 
	<dependencies>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>3.8.1</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>log4j</groupId>
			<artifactId>log4j</artifactId>
			<version>1.2.15</version>
		</dependency>
		<dependency>
			<groupId>org.apache.beam</groupId>
			<artifactId>beam-sdks-java-core</artifactId>
			<version>2.0.0</version>
		</dependency>
		<dependency>
			<groupId>org.apache.beam</groupId>
			<artifactId>beam-runners-direct-java</artifactId>
			<version>2.0.0</version>
		</dependency>
 
		<dependency>
			<groupId>org.apache.beam</groupId>
			<artifactId>beam-sdks-java-io-jdbc</artifactId>
			<version>2.0.0</version>
		</dependency>
 
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<version>5.1.38</version>
		</dependency>
 
	</dependencies>
	<build>
		<plugins>
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-compiler-plugin</artifactId>
				<configuration>
					<source>1.7</source>
					<target>1.7</target>
					<encoding>UTF8</encoding>
					<fork>fasle</fork>
					<meminitial>1024m</meminitial>
					<maxmem>2024m</maxmem>
				</configuration>
			</plugin>
 
 
			<!-- The configuration of maven-jar-plugin -->
			<plugin>
				<groupId>org.apache.maven.plugins</groupId>
				<artifactId>maven-jar-plugin</artifactId>
				<!-- The configuration of the plugin -->
				<configuration>
					<!-- Configuration of the archiver -->
					<archive>
						<!-- 生成的jar中，不要包含pom.xml和pom.properties这两个文件 -->
						<addMavenDescriptor>false</addMavenDescriptor>
						<!-- Manifest specific configuration -->
						<manifest>
							<!-- 是否要把第三方jar放到manifest的classpath中 -->
							<addClasspath>true</addClasspath>
							<!-- 生成的manifest中classpath的前缀，因为要把第三方jar放到lib目录下，所以classpath的前缀是lib/ -->
							<classpathPrefix>./</classpathPrefix>
							<!-- 应用的main class -->
							<mainClass>william_lib.wordcount</mainClass>
						</manifest>
					</archive>
				</configuration>
			</plugin>
		</plugins>
	</build>
</project>

package william_lib.wordcount;
 
import org.apache.beam.runners.direct.DirectRunner;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.ParDo;
import org.apache.beam.sdk.transforms.SimpleFunction;
import org.apache.beam.sdk.transforms.join.CoGbkResult;
import org.apache.beam.sdk.transforms.join.CoGroupByKey;
import org.apache.beam.sdk.transforms.join.KeyedPCollectionTuple;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.TupleTag;
 
/**
 *
 * <pre>
 * 2个本地数据集，以指定字段做join操作
 * </pre>
 *
 * @author William_JM
 * @date 2017年12月13日
 *
 */
public class TestJoin {
  @SuppressWarnings("serial")
  public static void main(String[] args) {
    // 管道工厂创建需求管道
    PipelineOptions options = PipelineOptionsFactory.create();
    // 显式指定PipelineRunner：DirectRunner（Local模式）
    options.setRunner(DirectRunner.class);
    // 装置管道
    Pipeline pipeline = Pipeline.create(options);
    // 读取用户信息数据集
    final PCollection<KV<String, String>> userInfoCollection =
        pipeline.apply(TextIO.read().from(args[0])).apply("userInfoCollection",
            MapElements.via(new SimpleFunction<String, KV<String, String>>() {
              @Override
              public KV<String, String> apply(String input) {
                // line format example:account|username
                String[] values = input.split("\\|");
                return KV.of(values[0], values[1]);
              }
            }));
    // 业务订单数据集
    final PCollection<KV<String, String>> orderCollection =
        pipeline.apply(TextIO.read().from(args[1])).apply("orderCollection",
            MapElements.via(new SimpleFunction<String, KV<String, String>>() {
              @Override
              public KV<String, String> apply(String input) {
                // line format example: orderId|orderName|description
                String[] values = input.split("\\|");
                return KV.of(values[0], values[1]);
              }
 
            }));
    final TupleTag<String> userInfoTag = new TupleTag<String>();
    final TupleTag<String> orderTag = new TupleTag<String>();
    // 通过 beam提供CoGroupByKey实现对2组关系数据集（key/value）的join操作
    // beam SDK为确保数据类型一致，强制数据集压入KeyedPCollectionTuple
    final PCollection<KV<String, CoGbkResult>> cogrouppedCollection =
        KeyedPCollectionTuple.of(userInfoTag, userInfoCollection).and(orderTag, orderCollection)
            .apply(CoGroupByKey.<String>create());
    final PCollection<KV<String, String>> finalResultCollection = cogrouppedCollection.apply(
        "finalResultCollection", ParDo.of(new DoFn<KV<String, CoGbkResult>, KV<String, String>>() {
          @ProcessElement
          public void processElement(ProcessContext pc) {
            KV<String, CoGbkResult> e = pc.element();
            String account = e.getKey();
            String username = e.getValue().getOnly(userInfoTag);
            for (String order : pc.element().getValue().getAll(orderTag)) {
              pc.output(KV.of(account, username + "\t" + order));
            }
          }
        }));
    // 结果保存数据集
    PCollection<String> formattedResults = finalResultCollection.apply("formattedResults",
        ParDo.of(new DoFn<KV<String, String>, String>() {
          @ProcessElement
          public void processElement(ProcessContext c) {
            c.output(c.element().getKey() + " : " + c.element().getValue());
          }
        }));
    formattedResults.apply(TextIO.write().to("joinedResults"));
    pipeline.run().waitUntilFinish();
  }
}