vpruthi/doc-exports
1
0
Fork 0
forked from docs/doc-exports
Code Pull Requests Activity
doc-exports/docs/mrs/component-operation-guide-lts/mrs_01_24247.html
Yang, Tong 3f5759eed2 MRS comp-lts 2.0.38.SP20 version 
Reviewed-by: Hasko, Vladimir <vladimir.hasko@t-systems.com>
Co-authored-by: Yang, Tong <yangtong2@huawei.com>
Co-committed-by: Yang, Tong <yangtong2@huawei.com>
2023-01-19 17:08:45 +00:00

238 lines
32 KiB
HTML
Raw Blame History

<a name="mrs_01_24247"></a><a name="mrs_01_24247"></a>
<h1 class="topictitle1">Interconnecting FlinkServer with HDFS</h1>
<div id="body8662426"><div class="section" id="mrs_01_24247__en-us_topic_0000001219029313_section11958749589"><h4 class="sectiontitle">Scenario</h4><p id="mrs_01_24247__en-us_topic_0000001219029313_p2074905315912">This section describes the data definition language (DDL) of HDFS as a sink table, as well as the WITH parameters and example code for creating a sink table, and provides guidance on how to perform operations on the FlinkServer job management page.</p>
<p id="mrs_01_24247__en-us_topic_0000001219029313_p13722011215">If your Kafka cluster is in security mode, the following example SQL statements can be used.</p>
</div>
<div class="section" id="mrs_01_24247__en-us_topic_0000001219029313_section14863102919913"><h4 class="sectiontitle">Prerequisites</h4><ul id="mrs_01_24247__ul14961152103613"><li id="mrs_01_24247__li196116525365">The HDFS, Yarn, and Flink services have been installed in a cluster.</li><li id="mrs_01_24247__li1961125211365">The client that contains the HDFS service has been installed, for example, in the <strong id="mrs_01_24247__en-us_topic_0000001219029313_b12934145535914">/opt/Bigdata/client</strong> directory.</li><li id="mrs_01_24247__li89611452153611">You have created a user with <strong id="mrs_01_24247__en-us_topic_0000001219029313_b20367458125914">FlinkServer Admin Privilege</strong>, for example, <strong id="mrs_01_24247__en-us_topic_0000001219029313_b236825810597">flink_admin</strong>, to access the Flink web UI. For details, see <a href="mrs_01_24049.html">Authentication Based on Users and Roles</a>.</li></ul>
</div>
<div class="section" id="mrs_01_24247__en-us_topic_0000001219029313_section1467216391592"><h4 class="sectiontitle">Procedure</h4><ol id="mrs_01_24247__en-us_topic_0000001219029313_ol17755531161917"><li id="mrs_01_24247__en-us_topic_0000001219029313_li12826131220399"><span>Log in to Manager as user <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1858463618432">flink_admin</strong> and choose <strong id="mrs_01_24247__en-us_topic_0000001219029313_b197164610219">Cluster</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1474461421">Services</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b98946125">Flink</strong>. In the <strong id="mrs_01_24247__en-us_topic_0000001219029313_b980461325">Basic Information</strong> area, click the link on the right of <strong id="mrs_01_24247__en-us_topic_0000001219029313_b487462023">Flink WebUI</strong> to access the Flink web UI.</span></li><li id="mrs_01_24247__en-us_topic_0000001219029313_li127091785297"><span>Create a Flink SQL job by referring to <a href="mrs_01_24024.html#mrs_01_24024__en-us_topic_0000001173470782_section1746418521537">Creating a Job</a>. On the job development page, configure the job parameters as follows and start the job.</span><p><div class="p" id="mrs_01_24247__en-us_topic_0000001219029313_p19976131517297">Select <strong id="mrs_01_24247__en-us_topic_0000001219029313_b167573519311">Enable CheckPoint</strong> in <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1976245113110">Running Parameter</strong> and set <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1476315153120">Time Interval (ms)</strong> to <strong id="mrs_01_24247__en-us_topic_0000001219029313_b17763451163119">60000</strong>.<pre class="screen" id="mrs_01_24247__en-us_topic_0000001219029313_screen12969195311266">CREATE TABLE kafka_table (
user_id STRING,
order_amount DOUBLE,
log_ts TIMESTAMP(3),
WATERMARK FOR log_ts AS log_ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'user_source',
'properties.bootstrap.servers' = '<em id="mrs_01_24247__en-us_topic_0000001219029313_i161999568315">IP address of the Kafka broker instance</em>:<em id="mrs_01_24247__en-us_topic_0000001219029313_i3205056203110">Kafka port number</em>',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'csv',
--Ignore the CSV data that fails to be parsed.
'csv.ignore-parse-errors' = 'true' ,--If the data is in JSON format, set 'json.ignore-parse-errors' to true.
'properties.sasl.kerberos.service.name' = 'kafka',
'properties.security.protocol' = 'SASL_PLAINTEXT',
'properties.kerberos.domain.name' = 'hadoop.<em id="mrs_01_24247__i1620672124713">System domain name</em>'
);
CREATE TABLE fs_table (
user_id STRING,
order_amount DOUBLE,
dt STRING,
`hour` STRING
) PARTITIONED BY (dt, `hour`) WITH ( --Date-specific file partitioning
'connector'='filesystem',
'path'='hdfs:///sql/parquet',
'format'='parquet',
'sink.partition-commit.delay'='1 h',
'sink.partition-commit.policy.kind'='success-file'
);
-- streaming sql, insert into file system table
INSERT INTO fs_table SELECT user_id, order_amount, DATE_FORMAT(log_ts, 'yyyy-MM-dd'), DATE_FORMAT(log_ts, 'HH') FROM kafka_table;</pre>
</div>
<div class="note" id="mrs_01_24247__en-us_topic_0000001219029313_note167015421341"><img src="public_sys-resources/note_3.0-en-us.png"><span class="notetitle"> </span><div class="notebody"><p id="mrs_01_24247__en-us_topic_0000001219029313_p3718569187">Kafka port number</p>
<ul id="mrs_01_24247__ul330585217498"><li id="mrs_01_24247__li15431748164911">In security mode, the port number is the value of <strong id="mrs_01_24247__b673611250427">sasl.port</strong> (<strong id="mrs_01_24247__b17736325174216">21007</strong> by default).</li><li id="mrs_01_24247__li175451448174913">In non-security mode, the port is the value of <strong id="mrs_01_24247__b13650127124215">port</strong> (<strong id="mrs_01_24247__b196511527174212">9092</strong> by default). If the port number is set to <strong id="mrs_01_24247__b9801172914211">9092</strong>, set <strong id="mrs_01_24247__b188021729164219">allow.everyone.if.no.acl.found</strong> to <strong id="mrs_01_24247__b880212910426">true</strong>. The procedure is as follows:<p id="mrs_01_24247__en-us_topic_0000001219029313_p8835244171713">Log in to FusionInsight Manager and choose <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1882175163611">Cluster</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b88835515361">Services</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b288415510368">Kafka</strong>. On the displayed page, click <strong id="mrs_01_24247__en-us_topic_0000001219029313_b18842583618">Configurations</strong> and then <strong id="mrs_01_24247__en-us_topic_0000001219029313_b18885145113618">All Configurations</strong>, search for <strong id="mrs_01_24247__en-us_topic_0000001219029313_b38856533618">allow.everyone.if.no.acl.found</strong>, set its value to <strong id="mrs_01_24247__en-us_topic_0000001219029313_b168865593615">true</strong>, and click <strong id="mrs_01_24247__en-us_topic_0000001219029313_b78871057363">Save</strong>.</p>
</li></ul>
</div></div>
</p></li><li id="mrs_01_24247__en-us_topic_0000001219029313_li1546149103815"><span>On the job management page, check whether the job status is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b6237195357735">Running</strong>.</span></li><li id="mrs_01_24247__en-us_topic_0000001219029313_li20986165822912"><span>Execute the following commands to view the topic and write data to Kafka. For details, see <a href="mrs_01_0379.html">Managing Messages in Kafka Topics</a>.</span><p><p id="mrs_01_24247__en-us_topic_0000001219029313_p098645832911"><strong id="mrs_01_24247__en-us_topic_0000001219029313_b14986115882915">./kafka-topics.sh --list --zookeeper </strong><em id="mrs_01_24247__en-us_topic_0000001219029313_i16986258112917">IP address of the ZooKeeper quorumpeer instance</em>:<em id="mrs_01_24247__en-us_topic_0000001219029313_i898612580294">ZooKeeper port number</em><strong id="mrs_01_24247__en-us_topic_0000001219029313_b18986105802913">/kafka</strong></p>
<p id="mrs_01_24247__en-us_topic_0000001219029313_p11986758102916"><strong id="mrs_01_24247__en-us_topic_0000001219029313_b398645852911">sh kafka-console-producer.sh --broker-list</strong> <em id="mrs_01_24247__en-us_topic_0000001219029313_i1798618582293">IP address of the node where the Kafka instance locates</em><em id="mrs_01_24247__en-us_topic_0000001219029313_i10986858192915">:Kafka port number</em><strong id="mrs_01_24247__en-us_topic_0000001219029313_b1498655872918"> --topic</strong> <em id="mrs_01_24247__en-us_topic_0000001219029313_i0986115819299">Topic name</em> --<strong id="mrs_01_24247__en-us_topic_0000001219029313_b9524833173416">producer.config</strong> <em id="mrs_01_24247__en-us_topic_0000001219029313_i956317274911">Client directory</em>/<strong id="mrs_01_24247__en-us_topic_0000001219029313_b159261443133416">Kafka/kafka/config/producer.properties</strong></p>
<p id="mrs_01_24247__en-us_topic_0000001219029313_p19986258192916">For example, if the topic name is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b32831214125614">user_source</strong>, the script is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b99861058122913">sh kafka-console-producer.sh --broker-list</strong> <em id="mrs_01_24247__en-us_topic_0000001219029313_i1398610580291">IP address of the node where the Kafka instance locates</em><em id="mrs_01_24247__en-us_topic_0000001219029313_i1898615810294">:Kafka port number</em> <strong id="mrs_01_24247__en-us_topic_0000001219029313_b398613584297">--topic user_source</strong> --<strong id="mrs_01_24247__en-us_topic_0000001219029313_b93871756133411">producer.config </strong><strong id="mrs_01_24247__en-us_topic_0000001219029313_b116651462134">/opt/</strong><strong id="mrs_01_24247__en-us_topic_0000001219029313_b56671068134">Bigdata/</strong><strong id="mrs_01_24247__en-us_topic_0000001219029313_b1668176201312">client</strong><strong id="mrs_01_24247__en-us_topic_0000001219029313_b17830931201114">/</strong><strong id="mrs_01_24247__en-us_topic_0000001219029313_b183891356193416">Kafka/kafka/config/producer.properties</strong>.</p>
<div class="p" id="mrs_01_24247__en-us_topic_0000001219029313_p1498618582294">Enter the message content.<pre class="screen" id="mrs_01_24247__en-us_topic_0000001219029313_screen19986195822918">3,3333,"2021-09-10 14:00"
4,4444,"2021-09-10 14:01"</pre>
</div>
<p id="mrs_01_24247__en-us_topic_0000001219029313_p1698685862916">Press <strong id="mrs_01_24247__en-us_topic_0000001219029313_b184711081439">Enter</strong> to send the message.</p>
<div class="note" id="mrs_01_24247__en-us_topic_0000001219029313_note480514320381"><img src="public_sys-resources/note_3.0-en-us.png"><span class="notetitle"> </span><div class="notebody"><ul id="mrs_01_24247__en-us_topic_0000001219029313_ul4278192283814"><li id="mrs_01_24247__en-us_topic_0000001219029313_li024511411817">IP address of the ZooKeeper quorumpeer instance<p id="mrs_01_24247__en-us_topic_0000001219029313_p92456140814"><a name="mrs_01_24247__en-us_topic_0000001219029313_li024511411817"></a><a name="en-us_topic_0000001219029313_li024511411817"></a>To obtain IP addresses of all ZooKeeper quorumpeer instances, log in to FusionInsight Manager and choose <strong id="mrs_01_24247__en-us_topic_0000001219029313_b18736171511430">Cluster</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b12736215134313">Services</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1073771514431">ZooKeeper</strong>. On the displayed page, click <strong id="mrs_01_24247__en-us_topic_0000001219029313_b573731504310">Instance</strong> and view the IP addresses of all the hosts where the quorumpeer instances locate.</p>
</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li624515142816">Port number of the ZooKeeper client<p id="mrs_01_24247__en-us_topic_0000001219029313_p1724512142082"><a name="mrs_01_24247__en-us_topic_0000001219029313_li624515142816"></a><a name="en-us_topic_0000001219029313_li624515142816"></a>Log in to FusionInsight Manager and choose <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1034152912437">Cluster</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b153452924313">Service</strong> &gt; <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1134122994311">ZooKeeper</strong>. On the displayed page, click <strong id="mrs_01_24247__en-us_topic_0000001219029313_b9353296435">Configurations</strong> and check the value of <strong id="mrs_01_24247__en-us_topic_0000001219029313_b4351429164315">clientPort</strong>. The default value is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b117691038184316">24002</strong>.</p>
</li></ul>
</div></div>
</p></li><li id="mrs_01_24247__en-us_topic_0000001219029313_li6386217104817"><span>Run the following command to check whether data is written from the HDFS directory to the sink table:</span><p><p id="mrs_01_24247__en-us_topic_0000001219029313_p129234324012"><strong id="mrs_01_24247__en-us_topic_0000001219029313_b1771114616402">hdfs dfs -ls -R /sql/parquet</strong></p>
</p></li></ol>
</div>
<div class="section" id="mrs_01_24247__en-us_topic_0000001219029313_section18208302534"><h4 class="sectiontitle">Interconnecting Flink with HDFS Partitions</h4><ul id="mrs_01_24247__en-us_topic_0000001219029313_ul882751174211"><li id="mrs_01_24247__en-us_topic_0000001219029313_li14706225468">Customized partitioning<p id="mrs_01_24247__en-us_topic_0000001219029313_p1650712364613"><a name="mrs_01_24247__en-us_topic_0000001219029313_li14706225468"></a><a name="en-us_topic_0000001219029313_li14706225468"></a>Flink's file system supports partitions in the standard Hive format. You do not need to register partitions with a table catalog. Partitions are inferred based on the directory structure.</p>
<div class="p" id="mrs_01_24247__en-us_topic_0000001219029313_p127225167434">For example, a table that is partitioned based on the following directory is inferred to contain datetime and hour partitions.<pre class="screen" id="mrs_01_24247__en-us_topic_0000001219029313_screen28111430145313">path
└── datetime=2021-09-03
└── hour=11
├── part-0.parquet
├── part-1.parquet
└── hour=12
├── part-0.parquet
└── datetime=2021-09-24
└── hour=6
├── part-0.parquet</pre>
</div>
</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li2063154744516">Rolling policy of partition files<p id="mrs_01_24247__en-us_topic_0000001219029313_p173906485450"><a name="mrs_01_24247__en-us_topic_0000001219029313_li2063154744516"></a><a name="en-us_topic_0000001219029313_li2063154744516"></a>Data in the partition directories is split into part files. Each partition contains at least one part file, which is used to receive the data written by the subtask of the sink.</p>
<div class="p" id="mrs_01_24247__en-us_topic_0000001219029313_p1247122010436">The following parameters describe the rolling policies of partition files.
<div class="tablenoborder"><table cellpadding="4" cellspacing="0" summary="" id="mrs_01_24247__en-us_topic_0000001219029313_table78121130185314" frame="border" border="1" rules="all"><thead align="left"><tr id="mrs_01_24247__en-us_topic_0000001219029313_row8812173017539"><th align="left" class="cellrowborder" valign="top" width="26.55%" id="mcps1.3.4.2.2.2.1.1.5.1.1"><p id="mrs_01_24247__en-us_topic_0000001219029313_p2812133012531">Parameter</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="12.34%" id="mcps1.3.4.2.2.2.1.1.5.1.2"><p id="mrs_01_24247__en-us_topic_0000001219029313_p48124305539">Default Value</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="14.829999999999998%" id="mcps1.3.4.2.2.2.1.1.5.1.3"><p id="mrs_01_24247__en-us_topic_0000001219029313_p16812230185313">Type</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="46.28%" id="mcps1.3.4.2.2.2.1.1.5.1.4"><p id="mrs_01_24247__en-us_topic_0000001219029313_p1812113016533">Description</p>
</th>
</tr>
</thead>
<tbody><tr id="mrs_01_24247__en-us_topic_0000001219029313_row0812193095310"><td class="cellrowborder" valign="top" width="26.55%" headers="mcps1.3.4.2.2.2.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p9812193019531">sink.rolling-policy.file-size</p>
</td>
<td class="cellrowborder" valign="top" width="12.34%" headers="mcps1.3.4.2.2.2.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p128122301533">128 MB</p>
</td>
<td class="cellrowborder" valign="top" width="14.829999999999998%" headers="mcps1.3.4.2.2.2.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p081263015537">MemorySize</p>
</td>
<td class="cellrowborder" valign="top" width="46.28%" headers="mcps1.3.4.2.2.2.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p181263011531">Maximum size of a partition file before it is rolled.</p>
</td>
</tr>
<tr id="mrs_01_24247__en-us_topic_0000001219029313_row1681283015312"><td class="cellrowborder" valign="top" width="26.55%" headers="mcps1.3.4.2.2.2.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p481223085313">sink.rolling-policy.rollover-interval</p>
</td>
<td class="cellrowborder" valign="top" width="12.34%" headers="mcps1.3.4.2.2.2.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p381283010535">30 minutes</p>
</td>
<td class="cellrowborder" valign="top" width="14.829999999999998%" headers="mcps1.3.4.2.2.2.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1812133015316">Duration</p>
</td>
<td class="cellrowborder" valign="top" width="46.28%" headers="mcps1.3.4.2.2.2.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p8812730175319">Maximum duration that a partition file can stay open before it is rolled.</p>
</td>
</tr>
<tr id="mrs_01_24247__en-us_topic_0000001219029313_row1681214308536"><td class="cellrowborder" valign="top" width="26.55%" headers="mcps1.3.4.2.2.2.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p16812830195315">sink.rolling-policy.check-interval</p>
</td>
<td class="cellrowborder" valign="top" width="12.34%" headers="mcps1.3.4.2.2.2.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p168121630115319">1 minute</p>
</td>
<td class="cellrowborder" valign="top" width="14.829999999999998%" headers="mcps1.3.4.2.2.2.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p681233025314">Duration</p>
</td>
<td class="cellrowborder" valign="top" width="46.28%" headers="mcps1.3.4.2.2.2.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1681243075313">Interval for checking time-based rolling policies.</p>
</td>
</tr>
</tbody>
</table>
</div>
</div>
</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li360811713468">File merging<div class="p" id="mrs_01_24247__en-us_topic_0000001219029313_p1195615174464"><a name="mrs_01_24247__en-us_topic_0000001219029313_li360811713468"></a><a name="en-us_topic_0000001219029313_li360811713468"></a>File compression is supported, allowing applications to have a shorter checkpoint interval without generating a large number of files.<div class="note" id="mrs_01_24247__en-us_topic_0000001219029313_note10745155518430"><img src="public_sys-resources/note_3.0-en-us.png"><span class="notetitle"> </span><div class="notebody"><p id="mrs_01_24247__en-us_topic_0000001219029313_p18746955124320">Only files in a single checkpoint are compressed. That is, the number of generated files is at least the same as the number of checkpoints. Files are invisible before merged. They are visible after both the checkpoint and compression are complete. If file compression takes too much time, the checkpoint will be prolonged.</p>
</div></div>
<div class="tablenoborder"><table cellpadding="4" cellspacing="0" summary="" id="mrs_01_24247__en-us_topic_0000001219029313_table14814163015530" frame="border" border="1" rules="all"><thead align="left"><tr id="mrs_01_24247__en-us_topic_0000001219029313_row1781363065312"><th align="left" class="cellrowborder" valign="top" width="25.1%" id="mcps1.3.4.2.3.1.2.1.5.1.1"><p id="mrs_01_24247__en-us_topic_0000001219029313_p188131830105315">Parameter</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="11.06%" id="mcps1.3.4.2.3.1.2.1.5.1.2"><p id="mrs_01_24247__en-us_topic_0000001219029313_p18138307531">Default Value</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="13.930000000000001%" id="mcps1.3.4.2.3.1.2.1.5.1.3"><p id="mrs_01_24247__en-us_topic_0000001219029313_p188132307536">Type</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="49.91%" id="mcps1.3.4.2.3.1.2.1.5.1.4"><p id="mrs_01_24247__en-us_topic_0000001219029313_p381318306535">Description</p>
</th>
</tr>
</thead>
<tbody><tr id="mrs_01_24247__en-us_topic_0000001219029313_row28139307539"><td class="cellrowborder" valign="top" width="25.1%" headers="mcps1.3.4.2.3.1.2.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p781313301531">auto-compaction</p>
</td>
<td class="cellrowborder" valign="top" width="11.06%" headers="mcps1.3.4.2.3.1.2.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p68131130115319">false</p>
</td>
<td class="cellrowborder" valign="top" width="13.930000000000001%" headers="mcps1.3.4.2.3.1.2.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p118133305534">Boolean</p>
</td>
<td class="cellrowborder" valign="top" width="49.91%" headers="mcps1.3.4.2.3.1.2.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p11813330105315">Whether to enable automatic compression. Data will be written to temporary files. After a checkpoint is complete, the temporary files generated by the checkpoint are compressed. These temporary files are invisible before compression.</p>
</td>
</tr>
<tr id="mrs_01_24247__en-us_topic_0000001219029313_row1981443010534"><td class="cellrowborder" valign="top" width="25.1%" headers="mcps1.3.4.2.3.1.2.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1681323055311">compaction.file-size</p>
</td>
<td class="cellrowborder" valign="top" width="11.06%" headers="mcps1.3.4.2.3.1.2.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1581303065314">none</p>
</td>
<td class="cellrowborder" valign="top" width="13.930000000000001%" headers="mcps1.3.4.2.3.1.2.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p108131307533">MemorySize</p>
</td>
<td class="cellrowborder" valign="top" width="49.91%" headers="mcps1.3.4.2.3.1.2.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p7814203020539">Size of the target file to be compressed. The default value is the size of the file to be rolled.</p>
</td>
</tr>
</tbody>
</table>
</div>
</div>
</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li2759143216463">Partition commit<div class="p" id="mrs_01_24247__en-us_topic_0000001219029313_p19498193315462"><a name="mrs_01_24247__en-us_topic_0000001219029313_li2759143216463"></a><a name="en-us_topic_0000001219029313_li2759143216463"></a>After a file is written to a partition, for example, a partition is added to Hive metastore (HMS) or a <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1421819581495">_SUCCESS</strong> file is written to a directory, the downstream application needs to be notified. Triggers and policies are used to commit partition files.<ul id="mrs_01_24247__en-us_topic_0000001219029313_ul1682093014534"><li id="mrs_01_24247__en-us_topic_0000001219029313_li981820302535">Trigger parameters for committing partition files
<div class="tablenoborder"><table cellpadding="4" cellspacing="0" summary="" id="mrs_01_24247__en-us_topic_0000001219029313_table1781763095317" frame="border" border="1" rules="all"><thead align="left"><tr id="mrs_01_24247__en-us_topic_0000001219029313_row7817830185312"><th align="left" class="cellrowborder" valign="top" width="19.35%" id="mcps1.3.4.2.4.1.2.1.1.1.5.1.1"><p id="mrs_01_24247__en-us_topic_0000001219029313_p281603011536">Parameter</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="10.7%" id="mcps1.3.4.2.4.1.2.1.1.1.5.1.2"><p id="mrs_01_24247__en-us_topic_0000001219029313_p188167307537">Default Value</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="9.53%" id="mcps1.3.4.2.4.1.2.1.1.1.5.1.3"><p id="mrs_01_24247__en-us_topic_0000001219029313_p6816630205319">Type</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="60.419999999999995%" id="mcps1.3.4.2.4.1.2.1.1.1.5.1.4"><p id="mrs_01_24247__en-us_topic_0000001219029313_p281713035314">Description</p>
</th>
</tr>
</thead>
<tbody><tr id="mrs_01_24247__en-us_topic_0000001219029313_row1781713307532"><td class="cellrowborder" valign="top" width="19.35%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p681719304535">sink.partition-commit.trigger</p>
</td>
<td class="cellrowborder" valign="top" width="10.7%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p158171930125312">process-time</p>
</td>
<td class="cellrowborder" valign="top" width="9.53%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1481743085320">String</p>
</td>
<td class="cellrowborder" valign="top" width="60.419999999999995%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.4 "><ul id="mrs_01_24247__en-us_topic_0000001219029313_ul18173015110"><li id="mrs_01_24247__en-us_topic_0000001219029313_li1283065113">process-time: System time of the compute node. It does not need to extract the partition time or generate watermarks. If the current system time exceeds the system time generated when a partition is created plus the delay time, the partition should be submitted.</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li17213304514">partition-time: Time extracted from the partition. Watermarks are required. If the time for generating watermarks exceeds the time extracted from a partition plus the delay time, the partition should be submitted.</li></ul>
</td>
</tr>
<tr id="mrs_01_24247__en-us_topic_0000001219029313_row281716306533"><td class="cellrowborder" valign="top" width="19.35%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p10817173010539">sink.partition-commit.delay</p>
</td>
<td class="cellrowborder" valign="top" width="10.7%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1081733017533">0 s</p>
</td>
<td class="cellrowborder" valign="top" width="9.53%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p15817153013534">Duration</p>
</td>
<td class="cellrowborder" valign="top" width="60.419999999999995%" headers="mcps1.3.4.2.4.1.2.1.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p2817103025318">Partitions will not be committed before the delay time. If it is a daily partition, the value is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b5681171415472">1 d</strong>. If it is an hourly one, the value is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b20177155274817">1 h</strong>.</p>
</td>
</tr>
</tbody>
</table>
</div>
</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li28201530155317">Policy parameters for committing partition files
<div class="tablenoborder"><table cellpadding="4" cellspacing="0" summary="" id="mrs_01_24247__en-us_topic_0000001219029313_table6819143075320" frame="border" border="1" rules="all"><thead align="left"><tr id="mrs_01_24247__en-us_topic_0000001219029313_row681813095319"><th align="left" class="cellrowborder" valign="top" width="25.16%" id="mcps1.3.4.2.4.1.2.2.1.1.5.1.1"><p id="mrs_01_24247__en-us_topic_0000001219029313_p6818830135319">Parameter</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="9.66%" id="mcps1.3.4.2.4.1.2.2.1.1.5.1.2"><p id="mrs_01_24247__en-us_topic_0000001219029313_p0818153095316">Default Value</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="9.030000000000001%" id="mcps1.3.4.2.4.1.2.2.1.1.5.1.3"><p id="mrs_01_24247__en-us_topic_0000001219029313_p1281873095318">Type</p>
</th>
<th align="left" class="cellrowborder" valign="top" width="56.15%" id="mcps1.3.4.2.4.1.2.2.1.1.5.1.4"><p id="mrs_01_24247__en-us_topic_0000001219029313_p481816302539">Description</p>
</th>
</tr>
</thead>
<tbody><tr id="mrs_01_24247__en-us_topic_0000001219029313_row2819113019533"><td class="cellrowborder" valign="top" width="25.16%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p2818830155310">sink.partition-commit.policy.kind</p>
</td>
<td class="cellrowborder" valign="top" width="9.66%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p188181030195317">-</p>
</td>
<td class="cellrowborder" valign="top" width="9.030000000000001%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p19818173016534">String</p>
</td>
<td class="cellrowborder" valign="top" width="56.15%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p15818143018535">Policy for committing partitions:</p>
<ul id="mrs_01_24247__en-us_topic_0000001219029313_ul10924725325"><li id="mrs_01_24247__en-us_topic_0000001219029313_li189249253220"><strong id="mrs_01_24247__en-us_topic_0000001219029313_b634613145141">metastore</strong>: used to add partitions to metastore. Only Hive tables support the metastore policy. The file system manages partitions based on the directory structure.</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li175361531520"><strong id="mrs_01_24247__en-us_topic_0000001219029313_b142411817181418">success-file</strong>: used to add <strong id="mrs_01_24247__en-us_topic_0000001219029313_b1828984217525">success-file</strong> files to a directory.</li><li id="mrs_01_24247__en-us_topic_0000001219029313_li13292124218275">The two policies can be configured at the same time, that is,<strong id="mrs_01_24247__en-us_topic_0000001219029313_b9116123365311"> 'sink.partition-commit.policy.kind'='metastore,success-file'</strong>.</li></ul>
</td>
</tr>
<tr id="mrs_01_24247__en-us_topic_0000001219029313_row168191530175318"><td class="cellrowborder" valign="top" width="25.16%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p481903025315">sink.partition-commit.policy.class</p>
</td>
<td class="cellrowborder" valign="top" width="9.66%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p1781963015317">-</p>
</td>
<td class="cellrowborder" valign="top" width="9.030000000000001%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p18819113017534">String</p>
</td>
<td class="cellrowborder" valign="top" width="56.15%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p16143184216331">Class that implements partition commit policy interfaces.</p>
<p id="mrs_01_24247__en-us_topic_0000001219029313_p0819133012538">This parameter takes effect only in the customized submission policies.</p>
</td>
</tr>
<tr id="mrs_01_24247__en-us_topic_0000001219029313_row2819123015536"><td class="cellrowborder" valign="top" width="25.16%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.1 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p28191530185313">sink.partition-commit.success-file.name</p>
</td>
<td class="cellrowborder" valign="top" width="9.66%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.2 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p181919307537">_SUCCESS</p>
</td>
<td class="cellrowborder" valign="top" width="9.030000000000001%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.3 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p481943045314">String</p>
</td>
<td class="cellrowborder" valign="top" width="56.15%" headers="mcps1.3.4.2.4.1.2.2.1.1.5.1.4 "><p id="mrs_01_24247__en-us_topic_0000001219029313_p19819153012534">File name of the success-file partition commit policy. The default value is <strong id="mrs_01_24247__en-us_topic_0000001219029313_b14372831115620">_SUCCESS</strong>.</p>
</td>
</tr>
</tbody>
</table>
</div>
</li></ul>
</div>
</li></ul>
</div>
</div>
<div>
<div class="familylinks">
<div class="parentlink"><strong>Parent topic:</strong> <a href="mrs_01_24226.html">Interconnecting FlinkServer with External Components</a></div>
</div>
</div>
View Git Blame Copy Permalink