examples/TableExample.java at master · xinyuiscool/examples · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
packagecom.linkedin.beam.examples;

importcom.linkedin.beam.examples.config.ConfigForExamples;
importcom.linkedin.beam.values.CoGroupWithTable;
importcom.linkedin.beam.values.PCollectionWithTable;
importcom.linkedin.beam.values.PTable;
importcom.linkedin.beam.values.ROTable;
importcom.linkedin.beam.values.PReadOnlyTable;
importcom.linkedin.beam.values.RWTable;
importcom.linkedin.beam.values.PCollectionTableJoin;
importcom.linkedin.beam.values.TableContext;
importcom.linkedin.beam.io.BrooklinIO;
importcom.linkedin.beam.io.BrooklinIOConfig;
importcom.linkedin.beam.io.LiKafkaIO;
importcom.linkedin.beam.io.LiKafkaIOConfig;
importcom.linkedin.events.PageViewEvent;
importcom.linkedin.identity.Profile;
importcom.linkedin.identity.internal.InternalSetting;
importorg.apache.avro.generic.GenericRecord;
importorg.apache.beam.runners.samza.SamzaPipelineOptions;
importorg.apache.beam.sdk.Pipeline;
importorg.apache.beam.sdk.coders.AvroCoder;
importorg.apache.beam.sdk.coders.StringUtf8Coder;
importorg.apache.beam.sdk.transforms.DoFn;
importorg.apache.beam.sdk.transforms.ParDo;
importorg.apache.beam.sdk.transforms.Values;
importorg.apache.beam.sdk.transforms.WithKeys;
importorg.apache.beam.sdk.transforms.join.CoGbkResult;
importorg.apache.beam.sdk.transforms.join.CoGroupByKey;
importorg.apache.beam.sdk.transforms.join.KeyedPCollectionTuple;
importorg.apache.beam.sdk.values.KV;
importorg.apache.beam.sdk.values.PCollection;
importorg.apache.beam.sdk.values.TupleTag;
importorg.apache.beam.sdk.values.TypeDescriptors;
importorg.joda.time.Instant;

importstaticcom.linkedin.beam.io.BrooklinIO.ESPRESSO;
importstaticcom.linkedin.beam.io.LiKafkaIOConfig.ClusterName.TRACKING;

publicclassTableExample {

publicstaticvoidmain(String[] args) {
finalSamzaPipelineOptionspipelineOpts = ConfigForExamples.getSamzaPipelineOptions("table-example");
finalLiKafkaIOConfigkafkaConfig = ConfigForExamples.getLiKafkaIOConfig();
finalBrooklinIOConfigbrooklinConfig = ConfigForExamples.getBrooklinIOConfig();
finalPipelinepipeline = Pipeline.create(pipelineOpts);

// A Brooklin stream
finalPCollection<KV<String, InternalSetting>> internalSettings = pipeline
 .apply(BrooklinIO.<GenericRecord, InternalSetting>read()
 .withConnectorName(ESPRESSO)
 .withStream("beam-test-brooklin-internal-setting")
 .withTimestampFn(kv -> newInstant(kv.getValue().getLastModified()))
 .withConfig(brooklinConfig)
 .withoutMetadata())
 .apply(Values.create())
 .apply(WithKeys
 .of(setting -> setting.getMemberId().toString()));

//Local RocksDb Table
finalPTable<KV<String, InternalSetting>> settingsTable =
pipeline.apply(
RocksDbTable.readWrite()
 .withName("internalSettings")
 .withKeyCoder(StringUtf8Coder.of())
 .withValueCoder(AvroCoder.of(PageViewEvent.class))
 .withInput(internalSettings));

// Remote Db Table
finalPReadOnlyTable<KV<String, Profile>> profileTable =
pipeline.apply(
EspressoTable.readOnly()
 .withDb("isb")
 .withTable("profile"));

// main input
PCollection<KV<String, PageViewEvent>> pageView = pipeline
 .apply(LiKafkaIO.<PageViewEvent>read()
 .withTopic("PageViewEvent")
 .withConsumerConfig(kafkaConfig.getConsumerConfig(TRACKING))
 .withoutMetadata());

// main input process with table
pageView
 .apply(TableParDo
 .of(
newDoFn<KV<String, PageViewEvent>, String>() {
@ProcessElement
publicvoidprocessElement(ProcessContextc,
@TableContext.InjectTableContexttc) {
StringmemberId = c.element().getKey();

//table lookup
RWTable<String, InternalSetting> settings = tc.getTable(settingsTable);
InternalSettingis = settings.get(memberId);
c.output(is.getName().toString());
 }
 })
 .withTables(settingsTable));

// Use the convenient helper class to do the same thing
PCollection<String> result = PCollectionTableJoin.of(pageView, settingsTable)
 .into(TypeDescriptors.strings())
 .via((pv, setting) -> setting.getName().toString());
 }
}