blocks|key|1520761|text|在这种情况下，我遵循的一种方法是创建一个特殊的键，它是两个键的组合。在将读取数据转换为键值对时，我会将session_id$cookie_id输出为一个连接字符串。这里，$可以是任何不构成两个键的字符集的分隔符。分隔符也可以忽略。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|1520762|entityMap^0|0^^$0|@$1|2|3|4|5|6|7|D|8|@]|9|@]|A|$]]|$1|B|3|-4|5|6|7|E|8|@]|9|@]|A|$]]]|C|$]]

One approach that I follow in such situations is to create an ad hoc key which is the combination of two keys . 
Post reading data , while converting to key value pair , I would output session_id$cookie_id as a single concatenated string . Here $ can be any delimiter which does not forms the charset of two keys . Delimiter can be ignored also .

blocks|key|1520826|text|扩展用户9720010的答案。您可以通过将字段映射到session_id和cookie_id的组合来创建复合键。此模式在通用数据流用例模式博客中进行了解释。假设您使用的是BigQuery，则可以执行以下类似操作：|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|1520827|Pipeline+pipeline+=+Pipeline.create(options);

//+Create+tuple+tags+for+the+value+types+in+each+collection.
final+TupleTag<TableRow>+table1Tag+=+new+TupleTag<>();
final+TupleTag<TableRow>+table2Tag+=+new+TupleTag<>();

//+Transform+for+keying+table+rows+by+session_id+and+cookie_id
WithKeys<String,+TableRow>+sessionAndCookieKeys+=+WithKeys.of(
++++(TableRow+row)+->
++++++++String.format("%25s#%25s",
++++++++++++row.get("session_id"),
++++++++++++row.get("cookie_id")))
++++.withKeyType(TypeDescriptors.strings());

/*
+*+Steps:
+*++1)+Read+table+1's+rows
+*++2)+Read+table+2's+rows
+*++3)+Map+each+row+to+a+composite+key
+*++4)+Join+on+the+composite+key
+*++5)+Process+the+results
+*/
PCollection<KV<String,+TableRow>>+table1Rows+=+pipeline
++++.apply(
++++++++"ReadTable1",
++++++++BigQueryIO
++++++++++++.readTableRows()
++++++++++++.from(options.getTable1()))
++++.apply("WithKeys",+sessionAndCookieKeys);

PCollection<KV<String,+TableRow>>+table2Rows+=+pipeline
++++.apply(
++++++++"ReadTable2",
++++++++BigQueryIO
++++++++++++.readTableRows()
++++++++++++.from(options.getTable2()))
++++.apply("WithKeys",+sessionAndCookieKeys);

//Merge+collection+values+into+a+CoGbkResult+collection
PCollection<KV<String,+CoGbkResult>>+coGbkResult+=+KeyedPCollectionTuple
++++.of(table1Tag,+table1Rows)
++++.and(table2Tag,+table2Rows)
++++.apply("JoinOnSessionAndCookie",+CoGroupByKey.create());

//+Process+the+results
coGbkResult.apply(
++++"ProcessResults",+
++++ParDo.of(new+DoFn<KV<String,+CoGbkResult>,+Object>()+{
++++++@ProcessElement
++++++public+void+processElement(ProcessContext+context)+{
++++++++//+Do+something+here
++++++}
++++}));|code-block|syntax|javascript|1520828|entityMap|0|LINK|mutability|MUTABLE|url|https://cloud.google.com/blog/big-data/2017/08/guide-to-common-cloud-dataflow-use-case-patterns-part-2^0|Q|A|11|9|1X|2|0|0|0^^$0|@$1|2|3|4|5|6|7|S|8|@$9|T|A|U|B|C]|$9|V|A|W|B|C]]|D|@$9|X|A|Y|1|Z]]|E|$]]|$1|F|3|G|5|H|7|10|8|@]|D|@]|E|$I|J]]|$1|K|3|-4|5|6|7|11|8|@]|D|@]|E|$]]]|L|$M|$5|N|O|P|E|$Q|R]]]]

To expand on user9720010's answer. You can create a composite key by mapping the fields to a combination of <code>session_id</code> and <code>cookie_id</code>. This pattern is explained in the common Dataflow use-case patterns <a href="https://cloud.google.com/blog/big-data/2017/08/guide-to-common-cloud-dataflow-use-case-patterns-part-2" rel="noreferrer">blog</a>. Assuming you are using BigQuery, you can do something similar to the following:

<pre><code>Pipeline pipeline = Pipeline.create(options);

// Create tuple tags for the value types in each collection.
final TupleTag&lt;TableRow&gt; table1Tag = new TupleTag&lt;&gt;();
final TupleTag&lt;TableRow&gt; table2Tag = new TupleTag&lt;&gt;();

// Transform for keying table rows by session_id and cookie_id
WithKeys&lt;String, TableRow&gt; sessionAndCookieKeys = WithKeys.of(
 (TableRow row) -&gt;
 String.format("%s#%s",
 row.get("session_id"),
 row.get("cookie_id")))
 .withKeyType(TypeDescriptors.strings());

/*
 * Steps:
 * 1) Read table 1's rows
 * 2) Read table 2's rows
 * 3) Map each row to a composite key
 * 4) Join on the composite key
 * 5) Process the results
 */
PCollection&lt;KV&lt;String, TableRow&gt;&gt; table1Rows = pipeline
 .apply(
 "ReadTable1",
 BigQueryIO
 .readTableRows()
 .from(options.getTable1()))
 .apply("WithKeys", sessionAndCookieKeys);

PCollection&lt;KV&lt;String, TableRow&gt;&gt; table2Rows = pipeline
 .apply(
 "ReadTable2",
 BigQueryIO
 .readTableRows()
 .from(options.getTable2()))
 .apply("WithKeys", sessionAndCookieKeys);

//Merge collection values into a CoGbkResult collection
PCollection&lt;KV&lt;String, CoGbkResult&gt;&gt; coGbkResult = KeyedPCollectionTuple
 .of(table1Tag, table1Rows)
 .and(table2Tag, table2Rows)
 .apply("JoinOnSessionAndCookie", CoGroupByKey.create());

// Process the results
coGbkResult.apply(
 "ProcessResults", 
 ParDo.of(new DoFn&lt;KV&lt;String, CoGbkResult&gt;, Object&gt;() {
 @ProcessElement
 public void processElement(ProcessContext context) {
 // Do something here
 }
 }));
</code></pre>

I have two tables A and B. Both of them have the fields <code>session_id</code> and <code>cookie_id</code>.
How do i create a Joined table output joining A with B on <code>session_id</code>, <code>cookie_id</code> with the help of a Dataflow pipeline? <code>CoGroupByKey</code> method allows you to join on a single key. Couldn't find anything helpful in the documentation as well.

How to join BQ tables on two or more keys with Cloud Dataflow?

翻译质量差，导致语言生硬或混乱。

没有提供实际的解决方法或示例。

解答不清晰，无法理解或解决问题。

页面排版不美观，阅读体验差。

文章

问答

视频

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

我有两个表A和B。它们都有字段session_id和cookie_id。如何在session_id上创建一个连接A与B连接的表输出，如何在Dataflow管道的帮助下创建cookie_id？CoGroupByKey方法允许您在一个键上连接。在文档中也找不到任何有用的东西。

问如何使用Cloud连接两个或多个键上的烧烤表？
EN

Stack Overflow用户

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问如何使用Cloud连接两个或多个键上的烧烤表？EN