Skip to content

[FLINK-37781][docs] Add elasticsearch lookup join document #125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[FLINK-37781][docs] Add elasticsearch lookup join document #125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions docs/content.zh/docs/connectors/table/elasticsearch.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,6 +222,51 @@ CREATE TABLE myUserTable (
默认使用内置的 <code>'json'</code> 格式。更多详细信息，请参阅 <a href="{{< ref "docs/connectors/table/formats/overview" >}}">JSON Format</a> 页面。
</td>
</tr>
<tr>
<td><h5>lookup.cache</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">NONE</td>
<td><p>枚举类型</p>可选值: NONE, PARTIAL</td>
<td>维表的缓存策略。 目前支持 NONE（不缓存）和 PARTIAL（只在外部数据库中查找数据时缓存）。</td>
</tr>
<tr>
<td><h5>lookup.partial-cache.max-rows</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
<td>查找缓存的最大行数，超过这个值，最旧的行将过期。使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。</td>
</tr>
<tr>
<td><h5>lookup.partial-cache.expire-after-write</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Duration</td>
<td>在记录写入缓存后该记录的最大保留时间。
使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。</td>
</tr>
<tr>
<td><h5>lookup.partial-cache.expire-after-access</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Duration</td>
<td>在缓存中的记录被访问后该记录的最大保留时间。
使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。</td>
</tr>
<tr>
<td><h5>lookup.partial-cache.caching-missing-key</h5></td>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should be 'lookup.partial-cache.cache-missing-key'

<td>可选</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>是否缓存维表中不存在的键，默认为true。
使用该配置时 "lookup.cache" 必须设置为 "PARTIAL”。</td>
</tr>
<tr>
<td><h5>lookup.max-retries</h5></td>
<td>可选</td>
<td style="word-wrap: break-word;">3</td>
<td>Integer</td>
<td>查找数据库失败时的最大重试次数。</td>
</tr>
</tbody>
</table>

Expand Down Expand Up @@ -257,6 +302,21 @@ Elasticsearch sink 同时支持静态索引和动态索引。

**注意:** 使用当前系统时间生成的动态索引时， 对于 changelog 的流，无法保证同一主键对应的记录能产生相同的索引名, 因此使用基于系统时间的动态索引，只能支持 append only 的流。

### Lookup Cache

Elasticsearch 连接器可以用在时态表关联中作为一个可 lookup 的 source (又称为维表)，当前只支持同步的查找模式。

默认情况下，lookup cache 是未启用的，你可以将 `lookup.cache` 设置为 `PARTIAL` 参数来启用。

lookup cache 的主要目的是用于提高时态表关联 Elasticsearch 连接器的性能。默认情况下，lookup cache 不开启，所以所有请求都会发送到外部数据库。
当 lookup cache 被启用时，每个进程（即 TaskManager）将维护一个缓存。Flink 将优先查找缓存，只有当缓存未查找到时才向外部数据库发送请求，并使用返回的数据更新缓存。
当缓存命中最大缓存行 `lookup.partial-cache.max-rows` 或当行超过 `lookup.partial-cache.expire-after-write` 或 `lookup.partial-cache.expire-after-access` 指定的最大存活时间时，缓存中的行将被设置为已过期。
缓存中的记录可能不是最新的，用户可以将缓存记录超时设置为一个更小的值以获得更好的刷新数据，但这可能会增加发送到数据库的请求数。所以要做好吞吐量和正确性之间的平衡。

默认情况下，flink 会缓存主键的空查询结果，你可以通过将 `lookup.partial-cache.cache-missing-key` 设置为 false 来切换行为。

<a name="idempotent-writes"></a>

数据类型映射
----------------

Expand Down
65 changes: 65 additions & 0 deletions docs/content/docs/connectors/table/elasticsearch.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,58 @@ Connector Options
By default uses built-in <code>'json'</code> format. Please refer to <a href="{{< ref "docs/connectors/table/formats/overview" >}}">JSON Format</a> page for more details.
</td>
</tr>
<tr>
<td><h5>lookup.cache</h5></td>
<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">NONE</td>
<td><p>Enum</p>Possible values: NONE, PARTIAL</td>
<td>The cache strategy for the lookup table. Currently supports NONE (no caching) and PARTIAL (caching entries on lookup operation in external database).</td>
</tr>
<tr>
<td><h5>lookup.partial-cache.max-rows</h5></td>
<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Long</td>
<td>The max number of rows of lookup cache, over this value, the oldest rows will be expired.
"lookup.cache" must be set to "PARTIAL" to use this option.</td>
</tr>
<tr>
<td><h5>lookup.partial-cache.expire-after-write</h5></td>
<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Duration</td>
<td>The max time to live for each rows in lookup cache after writing into the cache
"lookup.cache" must be set to "PARTIAL" to use this option. </td>
</tr>
<tr>
<td><h5>lookup.partial-cache.expire-after-access</h5></td>
<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">(none)</td>
<td>Duration</td>
<td>The max time to live for each rows in lookup cache after accessing the entry in the cache.
"lookup.cache" must be set to "PARTIAL" to use this option. </td>
</tr>
<tr>
<td><h5>lookup.partial-cache.caching-missing-key</h5></td>
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should be 'lookup.partial-cache.cache-missing-key'

<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">true</td>
<td>Boolean</td>
<td>Whether to store an empty value into the cache if the lookup key doesn't match any rows in the table.
"lookup.cache" must be set to "PARTIAL" to use this option.</td>
</tr>
<tr>
<td><h5>lookup.max-retries</h5></td>
<td>optional</td>
<td>yes</td>
<td style="word-wrap: break-word;">3</td>
<td>Integer</td>
<td>The max retry times if lookup database failed.</td>
</tr>
</tbody>
</table>

Expand Down Expand Up @@ -280,6 +332,19 @@ When formatting the system time as a string, the time zone configured in the ses
**NOTE:** When using the dynamic index generated by the current system time, for changelog stream, there is no guarantee that the records with the same primary key can generate the same index name.
Therefore, the dynamic index based on the system time can only support append only stream.

### Lookup Cache

Elasticsearch connector can be used in temporal join as a lookup source (aka. dimension table). Currently, only sync lookup mode is supported.

By default, lookup cache is not enabled. You can enable it by setting `lookup.cache` to `PARTIAL`.

The lookup cache is used to improve performance of temporal join the Elasticsearch connector. By default, lookup cache is not enabled, so all the requests are sent to external database.
When lookup cache is enabled, each process (i.e. TaskManager) will hold a cache. Flink will lookup the cache first, and only send requests to external database when cache missing, and update cache with the rows returned.
The oldest rows in cache will be expired when the cache hit to the max cached rows `lookup.partial-cache.max-rows` or when the row exceeds the max time to live specified by `lookup.partial-cache.expire-after-write` or `lookup.partial-cache.expire-after-access`.
The cached rows might not be the latest, users can tune expiration options to a smaller value to have a better fresh data, but this may increase the number of requests send to database. So this is a balance between throughput and correctness.

By default, flink will cache the empty query result for a Primary key, you can toggle the behaviour by setting `lookup.partial-cache.cache-missing-key` to false.

Data Type Mapping
----------------

Expand Down