<!doctype html>
<html lang="en-US" data-theme="light">
  <head>
    <meta charset="utf-8" />
    <meta name="viewport" content="width=device-width,initial-scale=1" />
    <meta name="generator" content="VuePress 2.0.0-rc.9" />
    <meta name="theme" content="VuePress Theme Hope 2.0.0-rc.34" />
    <style>
      html {
        background: var(--bg-color, #fff);
      }

      html[data-theme="dark"] {
        background: var(--bg-color, #1d1e1f);
      }

      body {
        background: var(--bg-color);
      }
    </style>
    <script>
      const userMode = localStorage.getItem("vuepress-theme-hope-scheme");
      const systemDarkMode =
        window.matchMedia &&
        window.matchMedia("(prefers-color-scheme: dark)").matches;

      if (userMode === "dark" || (userMode !== "light" && systemDarkMode)) {
        document.documentElement.setAttribute("data-theme", "dark");
      }
    </script>
    <link rel="alternate" hreflang="zh-cn" href="https://iotdb.apache.org/zh/UserGuide/V1.2.x/User-Manual/Streaming.html"><meta property="og:url" content="https://iotdb.apache.org/UserGuide/V1.2.x/User-Manual/Streaming.html"><meta property="og:site_name" content="IoTDB Website"><meta property="og:title" content="IoTDB Stream Processing Framework"><meta property="og:description" content="IoTDB Stream Processing Framework The IoTDB stream processing framework allows users to implement customized stream processing logic, which can monitor and capture storage engin..."><meta property="og:type" content="article"><meta property="og:image" content="https://alioss.timecho.com/docs/img/pipe.png"><meta property="og:locale" content="en-US"><meta property="og:locale:alternate" content="zh-CN"><meta property="og:updated_time" content="2024-04-08T07:45:44.000Z"><meta property="article:modified_time" content="2024-04-08T07:45:44.000Z"><script type="application/ld+json">{"@context":"https://schema.org","@type":"Article","headline":"IoTDB Stream Processing Framework","image":["https://alioss.timecho.com/docs/img/pipe.png","https://alioss.timecho.com/docs/img/%E7%8A%B6%E6%80%81%E8%BF%81%E7%A7%BB%E5%9B%BE.png"],"dateModified":"2024-04-08T07:45:44.000Z","author":[]}</script><link rel="icon" href="/favicon.ico"><meta name="Description" content="Apache IoTDB: Time Series Database for IoT"><meta name="Keywords" content="TSDB, time series, time series database, IoTDB, IoT database, IoT data management,时序数据库, 时间序列管理, IoTDB, 物联网数据库, 实时数据库, 物联网数据管理, 物联网数据"><meta name="baidu-site-verification" content="wfKETzB3OT"><meta name="google-site-verification" content="mZWAoRY0yj_HAr-s47zHCGHzx5Ju-RVm5wDbPnwQYFo"><script type="text/javascript">
var _paq = window._paq = window._paq || [];
/* tracker methods like "setCustomDimension" should be called before "trackPageView" */
_paq.push(["setDoNotTrack", true]);
_paq.push(["disableCookies"]);
_paq.push(['trackPageView']);
_paq.push(['enableLinkTracking']);
(function() {
  var u="https://analytics.apache.org/";
  _paq.push(['setTrackerUrl', u+'matomo.php']);
  _paq.push(['setSiteId', '56']);
  var d=document, g=d.createElement('script'), s=d.getElementsByTagName('script')[0];
  g.async=true; g.src=u+'matomo.js'; s.parentNode.insertBefore(g,s);
})();
       </script><title>IoTDB Stream Processing Framework | IoTDB Website</title><meta name="description" content="IoTDB Stream Processing Framework The IoTDB stream processing framework allows users to implement customized stream processing logic, which can monitor and capture storage engin...">
    <link rel="preload" href="/assets/style-DnEHAOmf.css" as="style"><link rel="stylesheet" href="/assets/style-DnEHAOmf.css">
    <link rel="modulepreload" href="/assets/app-0jI7BR7G.js"><link rel="modulepreload" href="/assets/Streaming.html-DS4Yt-lF.js">
    
  </head>
  <body>
    <div id="app"><!--[--><!--[--><!--[--><span tabindex="-1"></span><a href="#main-content" class="vp-skip-link sr-only">Skip to main content</a><!--]--><!--[--><div class="theme-container has-toc"><!--[--><header id="navbar" class="vp-navbar hide-icon"><div class="vp-navbar-start"><button type="button" class="vp-toggle-sidebar-button" title="Toggle Sidebar"><span class="icon"></span></button><!--[--><!----><!--]--><!--[--><a class="route-link vp-brand" href="/"><img class="vp-nav-logo" src="/logo.png" alt><!----><span class="vp-site-name hide-in-pad">IoTDB Website</span></a><!--]--><!--[--><!----><!--]--></div><div class="vp-navbar-center"><!--[--><!----><!--]--><!--[--><!--]--><!--[--><!----><!--]--></div><div class="vp-navbar-end"><!--[--><!----><!--]--><!--[--><!--[--><div id="docsearch-container" style="display:none;"></div><div><button type="button" class="DocSearch DocSearch-Button" aria-label="Search"><span class="DocSearch-Button-Container"><svg width="20" height="20" class="DocSearch-Search-Icon" viewBox="0 0 20 20"><path d="M14.386 14.386l4.0877 4.0877-4.0877-4.0877c-2.9418 2.9419-7.7115 2.9419-10.6533 0-2.9419-2.9418-2.9419-7.7115 0-10.6533 2.9418-2.9419 7.7115-2.9419 10.6533 0 2.9419 2.9418 2.9419 7.7115 0 10.6533z" stroke="currentColor" fill="none" fill-rule="evenodd" stroke-linecap="round" stroke-linejoin="round"></path></svg><span class="DocSearch-Button-Placeholder">Search</span></span><span class="DocSearch-Button-Keys"><kbd class="DocSearch-Button-Key"><svg width="15" height="15" class="DocSearch-Control-Key-Icon"><path d="M4.505 4.496h2M5.505 5.496v5M8.216 4.496l.055 5.993M10 7.5c.333.333.5.667.5 1v2M12.326 4.5v5.996M8.384 4.496c1.674 0 2.116 0 2.116 1.5s-.442 1.5-2.116 1.5M3.205 9.303c-.09.448-.277 1.21-1.241 1.203C1 10.5.5 9.513.5 8V7c0-1.57.5-2.5 1.464-2.494.964.006 1.134.598 1.24 1.342M12.553 10.5h1.953" stroke-width="1.2" stroke="currentColor" fill="none" stroke-linecap="square"></path></svg></kbd><kbd class="DocSearch-Button-Key">K</kbd></span></button></div><!--]--><nav class="vp-nav-links"><div class="vp-nav-item hide-in-mobile"><div class="dropdown-wrapper"><button type="button" class="dropdown-title" aria-label="Documentation"><span class="title"><!---->Documentation</span><span class="arrow"></span><ul class="nav-dropdown"><li class="dropdown-item"><a class="route-link nav-link" href="/UserGuide/latest/QuickStart/QuickStart.html" aria-label="v1.3.x"><!---->v1.3.x<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/UserGuide/V1.2.x/QuickStart/QuickStart.html" aria-label="v1.2.x"><!---->v1.2.x<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/UserGuide/V1.1.x/QuickStart/QuickStart.html" aria-label="v1.1.x"><!---->v1.1.x<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/UserGuide/V1.0.x/QuickStart/QuickStart.html" aria-label="v1.0.x"><!---->v1.0.x<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/UserGuide/V0.13.x/QuickStart/QuickStart.html" aria-label="v0.13.x"><!---->v0.13.x<!----></a></li></ul></button></div></div><div class="vp-nav-item hide-in-mobile"><a href="https://cwiki.apache.org/confluence/display/IOTDB/System+Design" rel="noopener noreferrer" target="_blank" aria-label="Design" class="nav-link"><!---->Design<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></div><div class="vp-nav-item hide-in-mobile"><a class="route-link nav-link" href="/Download/" aria-label="Download"><!---->Download<!----></a></div><div class="vp-nav-item hide-in-mobile"><div class="dropdown-wrapper"><button type="button" class="dropdown-title" aria-label="Community"><span class="title"><!---->Community</span><span class="arrow"></span><ul class="nav-dropdown"><li class="dropdown-item"><a class="route-link nav-link" href="/Community/About.html" aria-label="About"><!---->About<!----></a></li><li class="dropdown-item"><a href="https://cwiki.apache.org/confluence/display/iotdb" rel="noopener noreferrer" target="_blank" aria-label="Wiki" class="nav-link"><!---->Wiki<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Community/Community-Project-Committers.html" aria-label="People"><!---->People<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Community/Community-Powered-By.html" aria-label="Powered  By"><!---->Powered  By<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Community/Materials.html" aria-label="Resources"><!---->Resources<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Community/Feedback.html" aria-label="Feedback"><!---->Feedback<!----></a></li></ul></button></div></div><div class="vp-nav-item hide-in-mobile"><div class="dropdown-wrapper"><button type="button" class="dropdown-title" aria-label="Development"><span class="title"><!---->Development</span><span class="arrow"></span><ul class="nav-dropdown"><li class="dropdown-item"><a class="route-link nav-link" href="/Development/VoteRelease.html" aria-label="How  to  vote"><!---->How  to  vote<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/HowToCommit.html" aria-label="How  to  Commit"><!---->How  to  Commit<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/HowToJoin.html" aria-label="Become  a  Contributor"><!---->Become  a  Contributor<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/Committer.html" aria-label="Become  a  Committer"><!---->Become  a  Committer<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/ContributeGuide.html" aria-label="ContributeGuide"><!---->ContributeGuide<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/HowtoContributeCode.html" aria-label="How  to  Contribute  Code"><!---->How  to  Contribute  Code<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/format-changelist.html" aria-label="Changelist  of  TsFile"><!---->Changelist  of  TsFile<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/Development/rpc-changelist.html" aria-label="Changelist  of  RPC"><!---->Changelist  of  RPC<!----></a></li></ul></button></div></div><div class="vp-nav-item hide-in-mobile"><div class="dropdown-wrapper"><button type="button" class="dropdown-title" aria-label="ASF"><span class="title"><!---->ASF</span><span class="arrow"></span><ul class="nav-dropdown"><li class="dropdown-item"><a href="https://www.apache.org/" rel="noopener noreferrer" target="_blank" aria-label="Foundation" class="nav-link"><!---->Foundation<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a href="https://www.apache.org/licenses/" rel="noopener noreferrer" target="_blank" aria-label="License" class="nav-link"><!---->License<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a href="https://www.apache.org/security/" rel="noopener noreferrer" target="_blank" aria-label="Security" class="nav-link"><!---->Security<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a href="https://www.apache.org/foundation/sponsorship.html" rel="noopener noreferrer" target="_blank" aria-label="Sponsorship" class="nav-link"><!---->Sponsorship<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a href="https://www.apache.org/foundation/thanks.html" rel="noopener noreferrer" target="_blank" aria-label="Thanks" class="nav-link"><!---->Thanks<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a href="https://www.apache.org/events/current-event" rel="noopener noreferrer" target="_blank" aria-label="Current  Events" class="nav-link"><!---->Current  Events<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li><li class="dropdown-item"><a href="https://privacy.apache.org/policies/privacy-policy-public.html" rel="noopener noreferrer" target="_blank" aria-label="Privacy" class="nav-link"><!---->Privacy<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></li></ul></button></div></div></nav><div class="vp-nav-item"><div class="dropdown-wrapper"><button type="button" class="dropdown-title" aria-label="Select language"><!--[--><svg xmlns="http://www.w3.org/2000/svg" class="icon i18n-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="i18n icon" style="width:1rem;height:1rem;vertical-align:middle;"><path d="M379.392 460.8 494.08 575.488l-42.496 102.4L307.2 532.48 138.24 701.44l-71.68-72.704L234.496 460.8l-45.056-45.056c-27.136-27.136-51.2-66.56-66.56-108.544h112.64c7.68 14.336 16.896 27.136 26.112 35.84l45.568 46.08 45.056-45.056C382.976 312.32 409.6 247.808 409.6 204.8H0V102.4h256V0h102.4v102.4h256v102.4H512c0 70.144-37.888 161.28-87.04 210.944L378.88 460.8zM576 870.4 512 1024H409.6l256-614.4H768l256 614.4H921.6l-64-153.6H576zM618.496 768h196.608L716.8 532.48 618.496 768z"></path></svg><!--]--><span class="arrow"></span><ul class="nav-dropdown"><li class="dropdown-item"><a class="route-link nav-link active" href="/UserGuide/V1.2.x/User-Manual/Streaming.html" aria-label="English"><!---->English<!----></a></li><li class="dropdown-item"><a class="route-link nav-link" href="/zh/UserGuide/V1.2.x/User-Manual/Streaming.html" aria-label="简体中文"><!---->简体中文<!----></a></li></ul></button></div></div><div class="vp-nav-item hide-in-mobile"><button type="button" id="appearance-switch"><svg xmlns="http://www.w3.org/2000/svg" class="icon auto-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="auto icon" style="display:none;"><path d="M512 992C246.92 992 32 777.08 32 512S246.92 32 512 32s480 214.92 480 480-214.92 480-480 480zm0-840c-198.78 0-360 161.22-360 360 0 198.84 161.22 360 360 360s360-161.16 360-360c0-198.78-161.22-360-360-360zm0 660V212c165.72 0 300 134.34 300 300 0 165.72-134.28 300-300 300z"></path></svg><svg xmlns="http://www.w3.org/2000/svg" class="icon dark-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="dark icon" style="display:none;"><path d="M524.8 938.667h-4.267a439.893 439.893 0 0 1-313.173-134.4 446.293 446.293 0 0 1-11.093-597.334A432.213 432.213 0 0 1 366.933 90.027a42.667 42.667 0 0 1 45.227 9.386 42.667 42.667 0 0 1 10.24 42.667 358.4 358.4 0 0 0 82.773 375.893 361.387 361.387 0 0 0 376.747 82.774 42.667 42.667 0 0 1 54.187 55.04 433.493 433.493 0 0 1-99.84 154.88 438.613 438.613 0 0 1-311.467 128z"></path></svg><svg xmlns="http://www.w3.org/2000/svg" class="icon light-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="light icon" style="display:block;"><path d="M952 552h-80a40 40 0 0 1 0-80h80a40 40 0 0 1 0 80zM801.88 280.08a41 41 0 0 1-57.96-57.96l57.96-58a41.04 41.04 0 0 1 58 58l-58 57.96zM512 752a240 240 0 1 1 0-480 240 240 0 0 1 0 480zm0-560a40 40 0 0 1-40-40V72a40 40 0 0 1 80 0v80a40 40 0 0 1-40 40zm-289.88 88.08-58-57.96a41.04 41.04 0 0 1 58-58l57.96 58a41 41 0 0 1-57.96 57.96zM192 512a40 40 0 0 1-40 40H72a40 40 0 0 1 0-80h80a40 40 0 0 1 40 40zm30.12 231.92a41 41 0 0 1 57.96 57.96l-57.96 58a41.04 41.04 0 0 1-58-58l58-57.96zM512 832a40 40 0 0 1 40 40v80a40 40 0 0 1-80 0v-80a40 40 0 0 1 40-40zm289.88-88.08 58 57.96a41.04 41.04 0 0 1-58 58l-57.96-58a41 41 0 0 1 57.96-57.96z"></path></svg></button></div><div class="vp-nav-item vp-action"><a class="vp-action-link" href="https://github.com/apache/iotdb" target="_blank" rel="noopener noreferrer" aria-label="GitHub"><svg xmlns="http://www.w3.org/2000/svg" class="icon github-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="github icon" style="width:1.25rem;height:1.25rem;vertical-align:middle;"><path d="M511.957 21.333C241.024 21.333 21.333 240.981 21.333 512c0 216.832 140.544 400.725 335.574 465.664 24.49 4.395 32.256-10.07 32.256-23.083 0-11.69.256-44.245 0-85.205-136.448 29.61-164.736-64.64-164.736-64.64-22.315-56.704-54.4-71.765-54.4-71.765-44.587-30.464 3.285-29.824 3.285-29.824 49.195 3.413 75.179 50.517 75.179 50.517 43.776 75.008 114.816 53.333 142.762 40.79 4.523-31.66 17.152-53.377 31.19-65.537-108.971-12.458-223.488-54.485-223.488-242.602 0-53.547 19.114-97.323 50.517-131.67-5.035-12.33-21.93-62.293 4.779-129.834 0 0 41.258-13.184 134.912 50.346a469.803 469.803 0 0 1 122.88-16.554c41.642.213 83.626 5.632 122.88 16.554 93.653-63.488 134.784-50.346 134.784-50.346 26.752 67.541 9.898 117.504 4.864 129.834 31.402 34.347 50.474 78.123 50.474 131.67 0 188.586-114.73 230.016-224.042 242.09 17.578 15.232 33.578 44.672 33.578 90.454v135.85c0 13.142 7.936 27.606 32.854 22.87C862.25 912.597 1002.667 728.747 1002.667 512c0-271.019-219.648-490.667-490.71-490.667z"></path></svg></a></div><!--]--><!--[--><!----><!--]--><button type="button" class="vp-toggle-navbar-button" aria-label="Toggle Navbar" aria-expanded="false" aria-controls="nav-screen"><span><span class="vp-top"></span><span class="vp-middle"></span><span class="vp-bottom"></span></span></button></div></header><!----><!--]--><!----><div class="toggle-sidebar-wrapper"><span class="arrow start"></span></div><aside id="sidebar" class="vp-sidebar"><!--[--><!----><!--]--><ul class="vp-sidebar-links"><li><section class="vp-sidebar-group"><p class="vp-sidebar-header"><!----><span class="vp-sidebar-title">IoTDB User Guide (V1.2.x)</span><!----></p><ul class="vp-sidebar-links"></ul></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">About IoTDB</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">Quick Start</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">Basic Concept</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">Deployment &amp; Maintenance</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable active" type="button"><!----><span class="vp-sidebar-title">User Manual</span><span class="vp-arrow down"></span></button><ul class="vp-sidebar-links"><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Syntax-Rule.html" aria-label="Syntax Rule"><!---->Syntax Rule<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Operate-Metadata.html" aria-label="Operate Metadata"><!---->Operate Metadata<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Write-Delete-Data.html" aria-label="Write Delete Data"><!---->Write Delete Data<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Query-Data.html" aria-label="Query Data"><!---->Query Data<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Operator-and-Expression.html" aria-label="Operator and Expression"><!---->Operator and Expression<!----></a></li><li><a class="route-link nav-link active vp-sidebar-link vp-sidebar-page active" href="/UserGuide/V1.2.x/User-Manual/Streaming.html" aria-label="Streaming"><!---->Streaming<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Data-Sync.html" aria-label="Data Sync"><!---->Data Sync<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Database-Programming.html" aria-label="Database Programming"><!---->Database Programming<!----></a></li><li><a class="route-link nav-link vp-sidebar-link vp-sidebar-page" href="/UserGuide/V1.2.x/User-Manual/Authority-Management.html" aria-label="Authority Management"><!---->Authority Management<!----></a></li></ul></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">Tools System</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">API</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">Ecosystem Integration</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">SQL Manual</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">FAQ</span><span class="vp-arrow end"></span></button><!----></section></li><li><section class="vp-sidebar-group"><button class="vp-sidebar-header clickable" type="button"><!----><span class="vp-sidebar-title">Reference</span><span class="vp-arrow end"></span></button><!----></section></li></ul><!--[--><!----><!--]--></aside><!--[--><main id="main-content" class="vp-page"><!--[--><!--[--><!----><!--]--><!----><nav class="vp-breadcrumb disable"></nav><div class="vp-page-title"><h1><!---->IoTDB Stream Processing Framework</h1><div class="page-info"><!----><!----><span class="page-date-info" aria-label="Writing Date"><svg xmlns="http://www.w3.org/2000/svg" class="icon calendar-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="calendar icon"><path d="M716.4 110.137c0-18.753-14.72-33.473-33.472-33.473-18.753 0-33.473 14.72-33.473 33.473v33.473h66.993v-33.473zm-334.87 0c0-18.753-14.72-33.473-33.473-33.473s-33.52 14.72-33.52 33.473v33.473h66.993v-33.473zm468.81 33.52H716.4v100.465c0 18.753-14.72 33.473-33.472 33.473a33.145 33.145 0 01-33.473-33.473V143.657H381.53v100.465c0 18.753-14.72 33.473-33.473 33.473a33.145 33.145 0 01-33.473-33.473V143.657H180.6A134.314 134.314 0 0046.66 277.595v535.756A134.314 134.314 0 00180.6 947.289h669.74a134.36 134.36 0 00133.94-133.938V277.595a134.314 134.314 0 00-133.94-133.938zm33.473 267.877H147.126a33.145 33.145 0 01-33.473-33.473c0-18.752 14.72-33.473 33.473-33.473h736.687c18.752 0 33.472 14.72 33.472 33.473a33.145 33.145 0 01-33.472 33.473z"></path></svg><span><!----></span><meta property="datePublished" content="2023-10-10T10:08:46.000Z"></span><span class="page-pageview-info" aria-label="Page views"><svg xmlns="http://www.w3.org/2000/svg" class="icon eye-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="eye icon"><path d="M992 512.096c0-5.76-.992-10.592-1.28-11.136-.192-2.88-1.152-8.064-2.08-10.816-.256-.672-.544-1.376-.832-2.08-.48-1.568-1.024-3.104-1.6-4.32C897.664 290.112 707.104 160 512 160c-195.072 0-385.632 130.016-473.76 322.592-1.056 2.112-1.792 4.096-2.272 5.856a55.512 55.512 0 00-.64 1.6c-1.76 5.088-1.792 8.64-1.632 7.744-.832 3.744-1.568 11.168-1.568 11.168-.224 2.272-.224 4.032.032 6.304 0 0 .736 6.464 1.088 7.808.128 1.824.576 4.512 1.12 6.976h-.032c.448 2.08 1.12 4.096 1.984 6.08.48 1.536.992 2.976 1.472 4.032C126.432 733.856 316.992 864 512 864c195.136 0 385.696-130.048 473.216-321.696 1.376-2.496 2.24-4.832 2.848-6.912.256-.608.48-1.184.672-1.728 1.536-4.48 1.856-8.32 1.728-8.32l-.032.032c.608-3.104 1.568-7.744 1.568-13.28zM512 672c-88.224 0-160-71.776-160-160s71.776-160 160-160 160 71.776 160 160-71.776 160-160 160z"></path></svg><span id="ArtalkPV" class="vp-pageview waline-pageview-count" data-path="/UserGuide/V1.2.x/User-Manual/Streaming.html" data-page-key="/UserGuide/V1.2.x/User-Manual/Streaming.html">...</span></span><span class="page-reading-time-info" aria-label="Reading Time"><svg xmlns="http://www.w3.org/2000/svg" class="icon timer-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="timer icon"><path d="M799.387 122.15c4.402-2.978 7.38-7.897 7.38-13.463v-1.165c0-8.933-7.38-16.312-16.312-16.312H256.33c-8.933 0-16.311 7.38-16.311 16.312v1.165c0 5.825 2.977 10.874 7.637 13.592 4.143 194.44 97.22 354.963 220.201 392.763-122.204 37.542-214.893 196.511-220.2 389.397-4.661 5.049-7.638 11.651-7.638 19.03v5.825h566.49v-5.825c0-7.379-2.849-13.981-7.509-18.9-5.049-193.016-97.867-351.985-220.2-389.527 123.24-37.67 216.446-198.453 220.588-392.892zM531.16 450.445v352.632c117.674 1.553 211.787 40.778 211.787 88.676H304.097c0-48.286 95.149-87.382 213.728-88.676V450.445c-93.077-3.107-167.901-81.297-167.901-177.093 0-8.803 6.99-15.793 15.793-15.793 8.803 0 15.794 6.99 15.794 15.793 0 80.261 63.69 145.635 142.01 145.635s142.011-65.374 142.011-145.635c0-8.803 6.99-15.793 15.794-15.793s15.793 6.99 15.793 15.793c0 95.019-73.789 172.82-165.96 177.093z"></path></svg><span>About 17 min</span><meta property="timeRequired" content="PT17M"></span><!----><!----></div><hr></div><div class="vp-toc-placeholder"><aside id="toc"><!--[--><!----><!--]--><div class="vp-toc-header">On This Page<button type="button" class="print-button" title="Print"><svg xmlns="http://www.w3.org/2000/svg" class="icon print-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="print icon"><path d="M819.2 364.8h-44.8V128c0-17.067-14.933-32-32-32H281.6c-17.067 0-32 14.933-32 32v236.8h-44.8C145.067 364.8 96 413.867 96 473.6v192c0 59.733 49.067 108.8 108.8 108.8h44.8V896c0 17.067 14.933 32 32 32h460.8c17.067 0 32-14.933 32-32V774.4h44.8c59.733 0 108.8-49.067 108.8-108.8v-192c0-59.733-49.067-108.8-108.8-108.8zM313.6 160h396.8v204.8H313.6V160zm396.8 704H313.6V620.8h396.8V864zM864 665.6c0 25.6-19.2 44.8-44.8 44.8h-44.8V588.8c0-17.067-14.933-32-32-32H281.6c-17.067 0-32 14.933-32 32v121.6h-44.8c-25.6 0-44.8-19.2-44.8-44.8v-192c0-25.6 19.2-44.8 44.8-44.8h614.4c25.6 0 44.8 19.2 44.8 44.8v192z"></path></svg></button><div class="arrow end"></div></div><div class="vp-toc-wrapper"><ul class="vp-toc-list"><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level2" href="#custom-stream-processing-plugin-development">Custom Stream Processing Plugin Development</a></li><li><ul class="vp-toc-list"><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#programming-development-dependencies">Programming development dependencies</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#event-driven-programming-model">Event-Driven Programming Model</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#custom-stream-processing-plugin-programming-interface-definition">Custom Stream Processing Plugin Programming Interface Definition</a></li><!----><!--]--></ul></li><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level2" href="#custom-stream-processing-plugin-management">Custom Stream Processing Plugin Management</a></li><li><ul class="vp-toc-list"><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#load-plugin-statement">Load Plugin Statement</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#delete-plugin-statement">Delete Plugin Statement</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#show-plugin-statement">Show Plugin Statement</a></li><!----><!--]--></ul></li><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level2" href="#system-pre-installed-stream-processing-plugin">System Pre-installed Stream Processing Plugin</a></li><li><ul class="vp-toc-list"><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#pre-built-extractor-plugin">Pre-built extractor Plugin</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#pre-built-processor-plugin">Pre-built Processor Plugin</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#pre-built-connector-plugin">Pre-built Connector Plugin</a></li><!----><!--]--></ul></li><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level2" href="#stream-processing-task-management">Stream Processing Task Management</a></li><li><ul class="vp-toc-list"><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#create-stream-processing-task">Create Stream Processing Task</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#start-stream-processing-task">Start Stream Processing Task</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#stop-stream-processing-task">Stop Stream Processing Task</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#delete-stream-processing-task">Delete Stream Processing Task</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#show-stream-processing-task">Show Stream Processing Task</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#stream-processing-task-running-status-migration">Stream Processing Task Running Status Migration</a></li><!----><!--]--></ul></li><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level2" href="#authority-management">Authority Management</a></li><li><ul class="vp-toc-list"><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#stream-processing-task">Stream Processing Task</a></li><!----><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level3" href="#stream-processing-task-plugin">Stream Processing Task Plugin</a></li><!----><!--]--></ul></li><!--]--><!--[--><li class="vp-toc-item"><a class="route-link vp-toc-link level2" href="#configure-parameters">Configure Parameters</a></li><!----><!--]--></ul><div class="vp-toc-marker" style="top:-1.7rem;"></div></div><!--[--><!----><!--]--></aside></div><!--[--><!----><!--]--><div class="theme-hope-content"><h1 id="iotdb-stream-processing-framework" tabindex="-1"><a class="header-anchor" href="#iotdb-stream-processing-framework"><span>IoTDB Stream Processing Framework</span></a></h1><p>The IoTDB stream processing framework allows users to implement customized stream processing logic, which can monitor and capture storage engine changes, transform changed data, and push transformed data outward.</p><p>We call <!---->. A stream processing task (Pipe) contains three subtasks:</p><ul><li>Extract</li><li>Process</li><li>Send (Connect)</li></ul><p>The stream processing framework allows users to customize the processing logic of three subtasks using Java language and process data in a UDF-like manner.<br> In a Pipe, the three subtasks mentioned above are executed and implemented by three types of plugins. Data flows through these three plugins sequentially for processing:<br> Pipe Extractor is used to extract data, Pipe Processor is used to process data, Pipe Connector is used to send data, and the final data will be sent to an external system.</p><p><strong>The model for a Pipe task is as follows:</strong></p><p><img src="https://alioss.timecho.com/docs/img/pipe.png" alt="pipe.png" loading="lazy"><br> A data stream processing task essentially describes the attributes of the Pipe Extractor, Pipe Processor, and Pipe Connector plugins.</p><p>Users can configure the specific attributes of these three subtasks declaratively using SQL statements. By combining different attributes, flexible data ETL (Extract, Transform, Load) capabilities can be achieved.</p><p>Using the stream processing framework, it is possible to build a complete data pipeline to fulfill various requirements such as <em>edge-to-cloud synchronization, remote disaster recovery, and read/write load balancing across multiple databases</em>.</p><h2 id="custom-stream-processing-plugin-development" tabindex="-1"><a class="header-anchor" href="#custom-stream-processing-plugin-development"><span>Custom Stream Processing Plugin Development</span></a></h2><h3 id="programming-development-dependencies" tabindex="-1"><a class="header-anchor" href="#programming-development-dependencies"><span>Programming development dependencies</span></a></h3><p>It is recommended to use Maven to build the project. Add the following dependencies in the <code>pom.xml</code> file. Please make sure to choose dependencies with the same version as the IoTDB server version.</p><div class="language-xml line-numbers-mode" data-ext="xml" data-title="xml"><pre class="language-xml"><code><span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>dependency</span><span class="token punctuation">&gt;</span></span>
    <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>groupId</span><span class="token punctuation">&gt;</span></span>org.apache.iotdb<span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>groupId</span><span class="token punctuation">&gt;</span></span>
    <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>artifactId</span><span class="token punctuation">&gt;</span></span>pipe-api<span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>artifactId</span><span class="token punctuation">&gt;</span></span>
    <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>version</span><span class="token punctuation">&gt;</span></span>1.2.1<span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>version</span><span class="token punctuation">&gt;</span></span>
    <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>scope</span><span class="token punctuation">&gt;</span></span>provided<span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>scope</span><span class="token punctuation">&gt;</span></span>
<span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>dependency</span><span class="token punctuation">&gt;</span></span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h3 id="event-driven-programming-model" tabindex="-1"><a class="header-anchor" href="#event-driven-programming-model"><span>Event-Driven Programming Model</span></a></h3><p>The design of user programming interfaces for stream processing plugins follows the principles of the event-driven programming model. In this model, events serve as the abstraction of data in the user programming interface. The programming interface is decoupled from the specific execution method, allowing the focus to be on describing how the system expects events (data) to be processed upon arrival.</p><p>In the user programming interface of stream processing plugins, events abstract the write operations of database data. Events are captured by the local stream processing engine and passed sequentially through the three stages of stream processing, namely Pipe Extractor, Pipe Processor, and Pipe Connector plugins. User logic is triggered and executed within these three plugins.</p><p>To accommodate both low-latency stream processing in low-load scenarios and high-throughput stream processing in high-load scenarios at the edge, the stream processing engine dynamically chooses the processing objects from operation logs and data files. Therefore, the user programming interface for stream processing requires the user to provide the handling logic for two types of events: TabletInsertionEvent for operation log write events and TsFileInsertionEvent for data file write events.</p><h4 id="tabletinsertionevent" tabindex="-1"><a class="header-anchor" href="#tabletinsertionevent"><span><strong>TabletInsertionEvent</strong></span></a></h4><p>The TabletInsertionEvent is a high-level data abstraction for user write requests, which provides the ability to manipulate the underlying data of the write request by providing a unified operation interface.</p><p>For different database deployments, the underlying storage structure corresponding to the operation log write event is different. For stand-alone deployment scenarios, the operation log write event is an encapsulation of write-ahead log (WAL) entries; for distributed deployment scenarios, the operation log write event is an encapsulation of individual node consensus protocol operation log entries.</p><p>For write operations generated by different write request interfaces of the database, the data structure of the request structure corresponding to the operation log write event is also different.IoTDB provides many write interfaces such as InsertRecord, InsertRecords, InsertTablet, InsertTablets, and so on, and each kind of write request uses a completely different serialisation method to generate a write request. completely different serialisation methods and generate different binary entries.</p><p>The existence of operation log write events provides users with a unified view of data operations, which shields the implementation differences of the underlying data structures, greatly reduces the programming threshold for users, and improves the ease of use of the functionality.</p><div class="language-java line-numbers-mode" data-ext="java" data-title="java"><pre class="language-java"><code><span class="token doc-comment comment">/** TabletInsertionEvent is used to define the event of data insertion. */</span>
<span class="token keyword">public</span> <span class="token keyword">interface</span> <span class="token class-name">TabletInsertionEvent</span> <span class="token keyword">extends</span> <span class="token class-name">Event</span> <span class="token punctuation">{</span>

  <span class="token doc-comment comment">/**
   * The consumer processes the data row by row and collects the results by RowCollector.
   *
   * <span class="token keyword">@return</span> <span class="token punctuation">{</span><span class="token keyword">@code</span> <span class="token code-section"><span class="token code language-java"><span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">&gt;</span></span></span></span><span class="token punctuation">}</span> a list of new TabletInsertionEvent contains the
   *     results collected by the RowCollector
   */</span>
  <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">&gt;</span></span> <span class="token function">processRowByRow</span><span class="token punctuation">(</span><span class="token class-name">BiConsumer</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Row</span><span class="token punctuation">,</span> <span class="token class-name">RowCollector</span><span class="token punctuation">&gt;</span></span> consumer<span class="token punctuation">)</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * The consumer processes the Tablet directly and collects the results by RowCollector.
   *
   * <span class="token keyword">@return</span> <span class="token punctuation">{</span><span class="token keyword">@code</span> <span class="token code-section"><span class="token code language-java"><span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">&gt;</span></span></span></span><span class="token punctuation">}</span> a list of new TabletInsertionEvent contains the
   *     results collected by the RowCollector
   */</span>
  <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">&gt;</span></span> <span class="token function">processTablet</span><span class="token punctuation">(</span><span class="token class-name">BiConsumer</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">Tablet</span><span class="token punctuation">,</span> <span class="token class-name">RowCollector</span><span class="token punctuation">&gt;</span></span> consumer<span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h4 id="tsfileinsertionevent" tabindex="-1"><a class="header-anchor" href="#tsfileinsertionevent"><span><strong>TsFileInsertionEvent</strong></span></a></h4><p>The TsFileInsertionEvent represents a high-level abstraction of the database&#39;s disk flush operation and is a collection of multiple TabletInsertionEvents.</p><p>IoTDB&#39;s storage engine is based on the LSM (Log-Structured Merge) structure. When data is written, the write operations are first flushed to log-structured files, while the written data is also stored in memory. When the memory reaches its capacity limit, a flush operation is triggered, converting the data in memory into a database file while deleting the previously written log entries. During the conversion from memory data to database file data, two compression processes, encoding compression and universal compression, are applied. As a result, the data in the database file occupies less space compared to the original data in memory.</p><p>In extreme network conditions, directly transferring data files is more cost-effective than transmitting individual write operations. It consumes lower network bandwidth and achieves faster transmission speed. However, there is no such thing as a free lunch. Performing calculations on data in the disk file incurs additional costs for file I/O compared to performing calculations directly on data in memory. Nevertheless, the coexistence of disk data files and memory write operations permits dynamic trade-offs and adjustments. It is based on this observation that the data file write event is introduced into the event model of the plugin.</p><p>In summary, the data file write event appears in the event stream of stream processing plugins in the following two scenarios:</p><ol><li><p>Historical data extraction: Before a stream processing task starts, all persisted write data exists in the form of TsFiles. When collecting historical data at the beginning of a stream processing task, the historical data is abstracted as TsFileInsertionEvent.</p></li><li><p>Real-time data extraction: During the execution of a stream processing task, if the speed of processing the log entries representing real-time operations is slower than the rate of write requests, the unprocessed log entries will be persisted to disk in the form of TsFiles. When these data are extracted by the stream processing engine, they are abstracted as TsFileInsertionEvent.</p></li></ol><div class="language-java line-numbers-mode" data-ext="java" data-title="java"><pre class="language-java"><code><span class="token doc-comment comment">/**
 * TsFileInsertionEvent is used to define the event of writing TsFile. Event data stores in disks,
 * which is compressed and encoded, and requires IO cost for computational processing.
 */</span>
<span class="token keyword">public</span> <span class="token keyword">interface</span> <span class="token class-name">TsFileInsertionEvent</span> <span class="token keyword">extends</span> <span class="token class-name">Event</span> <span class="token punctuation">{</span>

  <span class="token doc-comment comment">/**
   * The method is used to convert the TsFileInsertionEvent into several TabletInsertionEvents.
   *
   * <span class="token keyword">@return</span> <span class="token punctuation">{</span><span class="token keyword">@code</span> <span class="token code-section"><span class="token code language-java"><span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">&gt;</span></span></span></span><span class="token punctuation">}</span> the list of TabletInsertionEvent
   */</span>
  <span class="token class-name">Iterable</span><span class="token generics"><span class="token punctuation">&lt;</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">&gt;</span></span> <span class="token function">toTabletInsertionEvents</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h3 id="custom-stream-processing-plugin-programming-interface-definition" tabindex="-1"><a class="header-anchor" href="#custom-stream-processing-plugin-programming-interface-definition"><span>Custom Stream Processing Plugin Programming Interface Definition</span></a></h3><p>Based on the custom stream processing plugin programming interface, users can easily write data extraction plugins, data processing plugins, and data sending plugins, allowing the stream processing functionality to adapt flexibly to various industrial scenarios.</p><h4 id="data-extraction-plugin-interface" tabindex="-1"><a class="header-anchor" href="#data-extraction-plugin-interface"><span>Data Extraction Plugin Interface</span></a></h4><p>Data extraction is the first stage of the three-stage process of stream processing, which includes data extraction, data processing, and data sending. The data extraction plugin (PipeExtractor) serves as a bridge between the stream processing engine and the storage engine. It captures various data write events by listening to the behavior of the storage engine.</p><div class="language-java line-numbers-mode" data-ext="java" data-title="java"><pre class="language-java"><code><span class="token doc-comment comment">/**
 * PipeExtractor
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>PipeExtractor is responsible for capturing events from sources.
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>Various data sources can be supported by implementing different PipeExtractor classes.
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>The lifecycle of a PipeExtractor is as follows:
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>When a collaboration task is created, the KV pairs of `WITH EXTRACTOR` clause in SQL are
 *       parsed and the validation method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>
 *       will be called to validate the parameters.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Before the collaboration task starts, the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *       <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeExtractorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called
 *       to config the runtime behavior of the PipeExtractor.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Then the method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">start</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called to start the PipeExtractor.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>While the collaboration task is in progress, the method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">supply</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will
 *       be called to capture events from sources and then the events will be passed to the
 *       PipeProcessor.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>The method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called when the collaboration task is
 *       cancelled (the `DROP PIPE` command is executed).
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
 */</span>
<span class="token keyword">public</span> <span class="token keyword">interface</span> <span class="token class-name">PipeExtractor</span> <span class="token keyword">extends</span> <span class="token class-name">PipePlugin</span> <span class="token punctuation">{</span>

  <span class="token doc-comment comment">/**
   * This method is mainly used to validate <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeParameters</span></span><span class="token punctuation">}</span> and it is executed before <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeExtractorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">validator</span> the validator used to validate <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeParameters</span></span><span class="token punctuation">}</span>
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> if any parameter is not valid
   */</span>
  <span class="token keyword">void</span> <span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span> validator<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is mainly used to customize PipeExtractor. In this method, the user can do the
   * following things:
   *
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
   *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Use PipeParameters to parse key-value pair attributes entered by the user.
   *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Set the running configurations in PipeExtractorRuntimeConfiguration.
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
   *
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>This method is called after the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">parameters</span> used to parse the input parameters entered by the user
   * <span class="token keyword">@param</span> <span class="token parameter">configuration</span> used to set the required properties of the running PipeExtractor
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span> parameters<span class="token punctuation">,</span> <span class="token class-name">PipeExtractorRuntimeConfiguration</span> configuration<span class="token punctuation">)</span>
          <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * Start the extractor. After this method is called, events should be ready to be supplied by
   * <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">supply</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>. This method is called after <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeExtractorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">start</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * Supply single event from the extractor and the caller will send the event to the processor.
   * This method is called after <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeExtractor</span><span class="token punctuation">#</span><span class="token function">start</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@return</span> the event to be supplied. the event may be null if the extractor has no more events at
   *     the moment, but the extractor is still running for more events.
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token class-name">Event</span> <span class="token function">supply</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h4 id="data-processing-plugin-interface" tabindex="-1"><a class="header-anchor" href="#data-processing-plugin-interface"><span>Data Processing Plugin Interface</span></a></h4><p>Data processing is the second stage of the three-stage process of stream processing, which includes data extraction, data processing, and data sending. The data processing plugin (PipeProcessor) is primarily used for filtering and transforming the various events captured by the data extraction plugin (PipeExtractor).</p><div class="language-java line-numbers-mode" data-ext="java" data-title="java"><pre class="language-java"><code><span class="token doc-comment comment">/**
 * PipeProcessor
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>PipeProcessor is used to filter and transform the Event formed by the PipeExtractor.
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>The lifecycle of a PipeProcessor is as follows:
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>When a collaboration task is created, the KV pairs of `WITH PROCESSOR` clause in SQL are
 *       parsed and the validation method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>
 *       will be called to validate the parameters.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Before the collaboration task starts, the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *       <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeProcessorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called
 *       to config the runtime behavior of the PipeProcessor.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>While the collaboration task is in progress:
 *       <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
 *         <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>PipeExtractor captures the events and wraps them into three types of Event instances.
 *         <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>PipeProcessor processes the event and then passes them to the PipeConnector. The
 *             following 3 methods will be called: <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *             <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">process</span><span class="token punctuation">(</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">,</span> <span class="token class-name">EventCollector</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>, <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *             <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">process</span><span class="token punctuation">(</span><span class="token class-name">TsFileInsertionEvent</span><span class="token punctuation">,</span> <span class="token class-name">EventCollector</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> and <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *             <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">process</span><span class="token punctuation">(</span><span class="token class-name">Event</span><span class="token punctuation">,</span> <span class="token class-name">EventCollector</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>.
 *         <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>PipeConnector serializes the events into binaries and send them to sinks.
 *       <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>When the collaboration task is cancelled (the `DROP PIPE` command is executed), the <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *       <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span> <span class="token punctuation">}</span> method will be called.
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
 */</span>
<span class="token keyword">public</span> <span class="token keyword">interface</span> <span class="token class-name">PipeProcessor</span> <span class="token keyword">extends</span> <span class="token class-name">PipePlugin</span> <span class="token punctuation">{</span>

  <span class="token doc-comment comment">/**
   * This method is mainly used to validate <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeParameters</span></span><span class="token punctuation">}</span> and it is executed before <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeProcessorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">validator</span> the validator used to validate <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeParameters</span></span><span class="token punctuation">}</span>
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> if any parameter is not valid
   */</span>
  <span class="token keyword">void</span> <span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span> validator<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is mainly used to customize PipeProcessor. In this method, the user can do the
   * following things:
   *
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
   *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Use PipeParameters to parse key-value pair attributes entered by the user.
   *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Set the running configurations in PipeProcessorRuntimeConfiguration.
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
   *
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>This method is called after the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeProcessor</span><span class="token punctuation">#</span><span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called and before the beginning of the
   * events processing.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">parameters</span> used to parse the input parameters entered by the user
   * <span class="token keyword">@param</span> <span class="token parameter">configuration</span> used to set the required properties of the running PipeProcessor
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span> parameters<span class="token punctuation">,</span> <span class="token class-name">PipeProcessorRuntimeConfiguration</span> configuration<span class="token punctuation">)</span>
          <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is called to process the TabletInsertionEvent.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">tabletInsertionEvent</span> TabletInsertionEvent to be processed
   * <span class="token keyword">@param</span> <span class="token parameter">eventCollector</span> used to collect result events after processing
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">process</span><span class="token punctuation">(</span><span class="token class-name">TabletInsertionEvent</span> tabletInsertionEvent<span class="token punctuation">,</span> <span class="token class-name">EventCollector</span> eventCollector<span class="token punctuation">)</span>
          <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is called to process the TsFileInsertionEvent.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">tsFileInsertionEvent</span> TsFileInsertionEvent to be processed
   * <span class="token keyword">@param</span> <span class="token parameter">eventCollector</span> used to collect result events after processing
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">default</span> <span class="token keyword">void</span> <span class="token function">process</span><span class="token punctuation">(</span><span class="token class-name">TsFileInsertionEvent</span> tsFileInsertionEvent<span class="token punctuation">,</span> <span class="token class-name">EventCollector</span> eventCollector<span class="token punctuation">)</span>
          <span class="token keyword">throws</span> <span class="token class-name">Exception</span> <span class="token punctuation">{</span>
    <span class="token keyword">for</span> <span class="token punctuation">(</span><span class="token keyword">final</span> <span class="token class-name">TabletInsertionEvent</span> tabletInsertionEvent <span class="token operator">:</span>
            tsFileInsertionEvent<span class="token punctuation">.</span><span class="token function">toTabletInsertionEvents</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
      <span class="token function">process</span><span class="token punctuation">(</span>tabletInsertionEvent<span class="token punctuation">,</span> eventCollector<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>
  <span class="token punctuation">}</span>

  <span class="token doc-comment comment">/**
   * This method is called to process the Event.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">event</span> Event to be processed
   * <span class="token keyword">@param</span> <span class="token parameter">eventCollector</span> used to collect result events after processing
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">process</span><span class="token punctuation">(</span><span class="token class-name">Event</span> event<span class="token punctuation">,</span> <span class="token class-name">EventCollector</span> eventCollector<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h4 id="data-sending-plugin-interface" tabindex="-1"><a class="header-anchor" href="#data-sending-plugin-interface"><span>Data Sending Plugin Interface</span></a></h4><p>Data sending is the third stage of the three-stage process of stream processing, which includes data extraction, data processing, and data sending. The data sending plugin (PipeConnector) is responsible for sending the various events processed by the data processing plugin (PipeProcessor). It serves as the network implementation layer of the stream processing framework and should support multiple real-time communication protocols and connectors in its interface.</p><div class="language-java line-numbers-mode" data-ext="java" data-title="java"><pre class="language-java"><code><span class="token doc-comment comment">/**
 * PipeConnector
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>PipeConnector is responsible for sending events to sinks.
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>Various network protocols can be supported by implementing different PipeConnector classes.
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>The lifecycle of a PipeConnector is as follows:
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>When a collaboration task is created, the KV pairs of `WITH CONNECTOR` clause in SQL are
 *       parsed and the validation method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>
 *       will be called to validate the parameters.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Before the collaboration task starts, the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *       <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeConnectorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called
 *       to config the runtime behavior of the PipeConnector and the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *       <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">handshake</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called to create a connection with sink.
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>While the collaboration task is in progress:
 *       <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
 *         <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>PipeExtractor captures the events and wraps them into three types of Event instances.
 *         <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>PipeProcessor processes the event and then passes them to the PipeConnector.
 *         <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>PipeConnector serializes the events into binaries and send them to sinks. The
 *             following 3 methods will be called: <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *             <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">transfer</span><span class="token punctuation">(</span><span class="token class-name">TabletInsertionEvent</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>, <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *             <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">transfer</span><span class="token punctuation">(</span><span class="token class-name">TsFileInsertionEvent</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> and <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *             <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">transfer</span><span class="token punctuation">(</span><span class="token class-name">Event</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>.
 *       <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
 *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>When the collaboration task is cancelled (the `DROP PIPE` command is executed), the <span class="token punctuation">{</span><span class="token keyword">@link</span>
 *       <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">close</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span> <span class="token punctuation">}</span> method will be called.
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
 *
 * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>In addition, the method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">heartbeat</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> will be called periodically to check
 * whether the connection with sink is still alive. The method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">handshake</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span>
 * will be called to create a new connection with the sink when the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
 * <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">heartbeat</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> throws exceptions.
 */</span>
<span class="token keyword">public</span> <span class="token keyword">interface</span> <span class="token class-name">PipeConnector</span> <span class="token keyword">extends</span> <span class="token class-name">PipePlugin</span> <span class="token punctuation">{</span>

  <span class="token doc-comment comment">/**
   * This method is mainly used to validate <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeParameters</span></span><span class="token punctuation">}</span> and it is executed before <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeConnectorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">validator</span> the validator used to validate <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeParameters</span></span><span class="token punctuation">}</span>
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> if any parameter is not valid
   */</span>
  <span class="token keyword">void</span> <span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span> validator<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is mainly used to customize PipeConnector. In this method, the user can do the
   * following things:
   *
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>ul</span><span class="token punctuation">&gt;</span></span>
   *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Use PipeParameters to parse key-value pair attributes entered by the user.
   *   <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>li</span><span class="token punctuation">&gt;</span></span>Set the running configurations in PipeConnectorRuntimeConfiguration.
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;/</span>ul</span><span class="token punctuation">&gt;</span></span>
   *
   * <span class="token tag"><span class="token tag"><span class="token punctuation">&lt;</span>p</span><span class="token punctuation">&gt;</span></span>This method is called after the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">validate</span><span class="token punctuation">(</span><span class="token class-name">PipeParameterValidator</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called and before the method <span class="token punctuation">{</span><span class="token keyword">@link</span>
   * <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">handshake</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is called.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">parameters</span> used to parse the input parameters entered by the user
   * <span class="token keyword">@param</span> <span class="token parameter">configuration</span> used to set the required properties of the running PipeConnector
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span> parameters<span class="token punctuation">,</span> <span class="token class-name">PipeConnectorRuntimeConfiguration</span> configuration<span class="token punctuation">)</span>
          <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is used to create a connection with sink. This method will be called after the
   * method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">customize</span><span class="token punctuation">(</span><span class="token class-name">PipeParameters</span><span class="token punctuation">,</span> <span class="token class-name">PipeConnectorRuntimeConfiguration</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> is
   * called or will be called when the method <span class="token punctuation">{</span><span class="token keyword">@link</span> <span class="token reference"><span class="token class-name">PipeConnector</span><span class="token punctuation">#</span><span class="token function">heartbeat</span><span class="token punctuation">(</span><span class="token punctuation">)</span></span><span class="token punctuation">}</span> throws exceptions.
   *
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> if the connection is failed to be created
   */</span>
  <span class="token keyword">void</span> <span class="token function">handshake</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method will be called periodically to check whether the connection with sink is still
   * alive.
   *
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> if the connection dies
   */</span>
  <span class="token keyword">void</span> <span class="token function">heartbeat</span><span class="token punctuation">(</span><span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is used to transfer the TabletInsertionEvent.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">tabletInsertionEvent</span> TabletInsertionEvent to be transferred
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">PipeConnectionException</span></span> if the connection is broken
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">transfer</span><span class="token punctuation">(</span><span class="token class-name">TabletInsertionEvent</span> tabletInsertionEvent<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>

  <span class="token doc-comment comment">/**
   * This method is used to transfer the TsFileInsertionEvent.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">tsFileInsertionEvent</span> TsFileInsertionEvent to be transferred
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">PipeConnectionException</span></span> if the connection is broken
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">default</span> <span class="token keyword">void</span> <span class="token function">transfer</span><span class="token punctuation">(</span><span class="token class-name">TsFileInsertionEvent</span> tsFileInsertionEvent<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span> <span class="token punctuation">{</span>
    <span class="token keyword">for</span> <span class="token punctuation">(</span><span class="token keyword">final</span> <span class="token class-name">TabletInsertionEvent</span> tabletInsertionEvent <span class="token operator">:</span>
            tsFileInsertionEvent<span class="token punctuation">.</span><span class="token function">toTabletInsertionEvents</span><span class="token punctuation">(</span><span class="token punctuation">)</span><span class="token punctuation">)</span> <span class="token punctuation">{</span>
      <span class="token function">transfer</span><span class="token punctuation">(</span>tabletInsertionEvent<span class="token punctuation">)</span><span class="token punctuation">;</span>
    <span class="token punctuation">}</span>
  <span class="token punctuation">}</span>

  <span class="token doc-comment comment">/**
   * This method is used to transfer the Event.
   *
   * <span class="token keyword">@param</span> <span class="token parameter">event</span> Event to be transferred
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">PipeConnectionException</span></span> if the connection is broken
   * <span class="token keyword">@throws</span> <span class="token reference"><span class="token class-name">Exception</span></span> the user can throw errors if necessary
   */</span>
  <span class="token keyword">void</span> <span class="token function">transfer</span><span class="token punctuation">(</span><span class="token class-name">Event</span> event<span class="token punctuation">)</span> <span class="token keyword">throws</span> <span class="token class-name">Exception</span><span class="token punctuation">;</span>
<span class="token punctuation">}</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h2 id="custom-stream-processing-plugin-management" tabindex="-1"><a class="header-anchor" href="#custom-stream-processing-plugin-management"><span>Custom Stream Processing Plugin Management</span></a></h2><p>To ensure the flexibility and usability of user-defined plugins in production environments, the system needs to provide the capability to dynamically manage plugins. This section introduces the management statements for stream processing plugins, which enable the dynamic and unified management of plugins.</p><h3 id="load-plugin-statement" tabindex="-1"><a class="header-anchor" href="#load-plugin-statement"><span>Load Plugin Statement</span></a></h3><p>In IoTDB, to dynamically load a user-defined plugin into the system, you first need to implement a specific plugin class based on PipeExtractor, PipeProcessor, or PipeConnector. Then, you need to compile and package the plugin class into an executable jar file. Finally, you can use the loading plugin management statement to load the plugin into IoTDB.</p><p>The syntax of the loading plugin management statement is as follows:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">CREATE</span> PIPEPLUGIN <span class="token operator">&lt;</span>alias<span class="token operator">&gt;</span>
<span class="token keyword">AS</span> <span class="token operator">&lt;</span><span class="token keyword">Full</span> class name<span class="token operator">&gt;</span>
<span class="token keyword">USING</span> <span class="token operator">&lt;</span>URL <span class="token keyword">of</span> jar <span class="token keyword">file</span><span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><p>For example, if a user implements a data processing plugin with the fully qualified class name &quot;edu.tsinghua.iotdb.pipe.ExampleProcessor&quot; and packages it into a jar file, which is stored at &quot;<a href="https://example.com:8080/iotdb/pipe-plugin.jar" target="_blank" rel="noopener noreferrer">https://example.com:8080/iotdb/pipe-plugin.jar<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span></a>&quot;, and the user wants to use this plugin in the stream processing engine, marking the plugin as &quot;example&quot;. The creation statement for this data processing plugin is as follows:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">CREATE</span> PIPEPLUGIN example
<span class="token keyword">AS</span> <span class="token string">&#39;edu.tsinghua.iotdb.pipe.ExampleProcessor&#39;</span>
<span class="token keyword">USING</span> URI <span class="token string">&#39;&lt;https://example.com:8080/iotdb/pipe-plugin.jar&gt;&#39;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><h3 id="delete-plugin-statement" tabindex="-1"><a class="header-anchor" href="#delete-plugin-statement"><span>Delete Plugin Statement</span></a></h3><p>When user no longer wants to use a plugin and needs to uninstall the plugin from the system, you can use the Remove plugin statement as shown below.</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">DROP</span> PIPEPLUGIN <span class="token operator">&lt;</span>alias<span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><h3 id="show-plugin-statement" tabindex="-1"><a class="header-anchor" href="#show-plugin-statement"><span>Show Plugin Statement</span></a></h3><p>User can also view the plugin in the system on need. The statement to view plugin is as follows.</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">SHOW</span> PIPEPLUGINS
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><h2 id="system-pre-installed-stream-processing-plugin" tabindex="-1"><a class="header-anchor" href="#system-pre-installed-stream-processing-plugin"><span>System Pre-installed Stream Processing Plugin</span></a></h2><h3 id="pre-built-extractor-plugin" tabindex="-1"><a class="header-anchor" href="#pre-built-extractor-plugin"><span>Pre-built extractor Plugin</span></a></h3><h4 id="iotdb-extractor" tabindex="-1"><a class="header-anchor" href="#iotdb-extractor"><span>iotdb-extractor</span></a></h4><p>Function: Extract historical or realtime data inside IoTDB into pipe.</p><table><thead><tr><th>key</th><th>value</th><th>value range</th><th>required or optional with default</th></tr></thead><tbody><tr><td>extractor</td><td>iotdb-extractor</td><td>String: iotdb-extractor</td><td>required</td></tr><tr><td>extractor.pattern</td><td>path prefix for filtering time series</td><td>String: any time series prefix</td><td>optional: root</td></tr><tr><td>extractor.history.enable</td><td>whether to sync historical data</td><td>Boolean: true, false</td><td>optional: true</td></tr><tr><td>extractor.history.start-time</td><td>start of synchronizing historical data event time，Include start-time</td><td>Long: [Long.MIN_VALUE, Long.MAX_VALUE]</td><td>optional: Long.MIN_VALUE</td></tr><tr><td>extractor.history.end-time</td><td>end of synchronizing historical data event time，Include end-time</td><td>Long: [Long.MIN_VALUE, Long.MAX_VALUE]</td><td>optional: Long.MAX_VALUE</td></tr><tr><td>extractor.realtime.enable</td><td>Whether to sync realtime data</td><td>Boolean: true, false</td><td>optional: true</td></tr></tbody></table><blockquote><p>🚫 <strong>extractor.pattern Parameter Description</strong></p><ul><li><p>Pattern should use backquotes to modify illegal characters or illegal path nodes, for example, if you want to filter root.`a@b` or root.`123`, you should set the pattern to root.`a@b` or root.`123`（Refer specifically to <a href="https://iotdb.apache.org/zh/Download/#_1-0-%E7%89%88%E6%9C%AC%E4%B8%8D%E5%85%BC%E5%AE%B9%E7%9A%84%E8%AF%AD%E6%B3%95%E8%AF%A6%E7%BB%86%E8%AF%B4%E6%98%8E" target="_blank" rel="noopener noreferrer">Timing of single and double quotes and backquotes<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span></a>）</p></li><li><p>In the underlying implementation, when pattern is detected as root (default value), synchronization efficiency is higher, and any other format will reduce performance.</p></li><li><p>The path prefix does not need to form a complete path. For example, when creating a pipe with the parameter &#39;extractor.pattern&#39;=&#39;root.aligned.1&#39;:</p><ul><li>root.aligned.1TS</li><li>root.aligned.1TS.`1`</li><li>root.aligned.100TS</li></ul><p>the data will be synchronized;</p><ul><li>root.aligned.`1`</li><li>root.aligned.`123`</li></ul><p>the data will not be synchronized.</p></li></ul></blockquote><blockquote><p>❗️<strong>start-time, end-time parameter description of extractor.history</strong></p><ul><li>start-time, end-time should be in ISO format, such as 2011-12-03T10:15:30 or 2011-12-03T10:15:30+01:00</li></ul></blockquote><blockquote><p>✅ <strong>a piece of data from production to IoTDB contains two key concepts of time</strong></p><ul><li><strong>event time：</strong> the time when the data is actually produced (or the generation time assigned to the data by the data production system, which is a time item in the data point), also called the event time.</li><li><strong>arrival time：</strong> the time the data arrived in the IoTDB system.</li></ul><p>The out-of-order data we often refer to refers to data whose <strong>event time</strong> is far behind the current system time (or the maximum <strong>event time</strong> that has been dropped) when the data arrives. On the other hand, whether it is out-of-order data or sequential data, as long as they arrive newly in the system, their <strong>arrival time</strong> will increase with the order in which the data arrives at IoTDB.</p></blockquote><blockquote><p>💎 <strong>the work of iotdb-extractor can be split into two stages</strong></p><ol><li>Historical data extraction: All data with <strong>arrival time</strong> &lt; <strong>current system time</strong> when creating the pipe is called historical data</li><li>Realtime data extraction: All data with <strong>arrival time</strong> &gt;= <strong>current system time</strong> when the pipe is created is called realtime data</li></ol><p>The historical data transmission phase and the realtime data transmission phase are executed serially. Only when the historical data transmission phase is completed, the realtime data transmission phase is executed.**</p><p>Users can specify iotdb-extractor to:</p><ul><li>Historical data extraction（<code>&#39;extractor.history.enable&#39; = &#39;true&#39;</code>, <code>&#39;extractor.realtime.enable&#39; = &#39;false&#39;</code> ）</li><li>Realtime data extraction（<code>&#39;extractor.history.enable&#39; = &#39;false&#39;</code>, <code>&#39;extractor.realtime.enable&#39; = &#39;true&#39;</code> ）</li><li>Full data extraction（<code>&#39;extractor.history.enable&#39; = &#39;true&#39;</code>, <code>&#39;extractor.realtime.enable&#39; = &#39;true&#39;</code> ）</li><li>Disable simultaneous sets <code>extractor.history.enable</code> and <code>extractor.realtime.enable</code> to <code>false</code></li></ul></blockquote><h3 id="pre-built-processor-plugin" tabindex="-1"><a class="header-anchor" href="#pre-built-processor-plugin"><span>Pre-built Processor Plugin</span></a></h3><h4 id="do-nothing-processor" tabindex="-1"><a class="header-anchor" href="#do-nothing-processor"><span>do-nothing-processor</span></a></h4><p>Function: Do not do anything with the events passed in by the extractor.</p><table><thead><tr><th>key</th><th>value</th><th>value range</th><th>required or optional with default</th></tr></thead><tbody><tr><td>processor</td><td>do-nothing-processor</td><td>String: do-nothing-processor</td><td>required</td></tr></tbody></table><h3 id="pre-built-connector-plugin" tabindex="-1"><a class="header-anchor" href="#pre-built-connector-plugin"><span>Pre-built Connector Plugin</span></a></h3><h4 id="do-nothing-connector" tabindex="-1"><a class="header-anchor" href="#do-nothing-connector"><span>do-nothing-connector</span></a></h4><p>Function: Does not do anything with the events passed in by the processor.</p><table><thead><tr><th>key</th><th>value</th><th>value range</th><th>required or optional with default</th></tr></thead><tbody><tr><td>connector</td><td>do-nothing-connector</td><td>String: do-nothing-connector</td><td>required</td></tr></tbody></table><h2 id="stream-processing-task-management" tabindex="-1"><a class="header-anchor" href="#stream-processing-task-management"><span>Stream Processing Task Management</span></a></h2><h3 id="create-stream-processing-task" tabindex="-1"><a class="header-anchor" href="#create-stream-processing-task"><span>Create Stream Processing Task</span></a></h3><p>A stream processing task can be created using the <code>CREATE PIPE</code> statement, a sample SQL statement is shown below:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">CREATE</span> PIPE <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span> <span class="token comment">-- PipeId is the name that uniquely identifies the sync task</span>
<span class="token keyword">WITH</span> EXTRACTOR <span class="token punctuation">(</span>
  <span class="token comment">-- Default IoTDB Data Extraction Plugin</span>
  <span class="token string">&#39;extractor&#39;</span>                    <span class="token operator">=</span> <span class="token string">&#39;iotdb-extractor&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Path prefix, only data that can match the path prefix will be extracted for subsequent processing and delivery</span>
  <span class="token string">&#39;extractor.pattern&#39;</span>            <span class="token operator">=</span> <span class="token string">&#39;root.timecho&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Whether to extract historical data</span>
  <span class="token string">&#39;extractor.history.enable&#39;</span>     <span class="token operator">=</span> <span class="token string">&#39;true&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Describes the time range of the historical data being extracted, indicating the earliest possible time</span>
  <span class="token string">&#39;extractor.history.start-time&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;2011.12.03T10:15:30+01:00&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Describes the time range of the extracted historical data, indicating the latest time</span>
  <span class="token string">&#39;extractor.history.end-time&#39;</span>   <span class="token operator">=</span> <span class="token string">&#39;2022.12.03T10:15:30+01:00&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Whether to extract realtime data</span>
  <span class="token string">&#39;extractor.realtime.enable&#39;</span>    <span class="token operator">=</span> <span class="token string">&#39;true&#39;</span><span class="token punctuation">,</span>
<span class="token punctuation">)</span>
<span class="token keyword">WITH</span> PROCESSOR <span class="token punctuation">(</span>
  <span class="token comment">-- Default data processing plugin, means no processing</span>
  <span class="token string">&#39;processor&#39;</span>                    <span class="token operator">=</span> <span class="token string">&#39;do-nothing-processor&#39;</span><span class="token punctuation">,</span>
<span class="token punctuation">)</span>
<span class="token keyword">WITH</span> CONNECTOR <span class="token punctuation">(</span>
  <span class="token comment">-- IoTDB data sending plugin with target IoTDB</span>
  <span class="token string">&#39;connector&#39;</span>                    <span class="token operator">=</span> <span class="token string">&#39;iotdb-thrift-connector&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Data service for one of the DataNode nodes on the target IoTDB ip</span>
  <span class="token string">&#39;connector.ip&#39;</span>                 <span class="token operator">=</span> <span class="token string">&#39;127.0.0.1&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Data service port of one of the DataNode nodes of the target IoTDB</span>
  <span class="token string">&#39;connector.port&#39;</span>               <span class="token operator">=</span> <span class="token string">&#39;6667&#39;</span><span class="token punctuation">,</span>
<span class="token punctuation">)</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><p><strong>To create a stream processing task it is necessary to configure the PipeId and the parameters of the three plugin sections:</strong></p><table><thead><tr><th>configuration item</th><th>description</th><th>Required or not</th><th>default implementation</th><th>Default implementation description</th><th>Whether to allow custom implementations</th></tr></thead><tbody><tr><td>pipeId</td><td>Globally uniquely identifies the name of a sync task</td><td><!----></td><td>-</td><td>-</td><td>-</td></tr><tr><td>extractor</td><td>pipe Extractor plugin, for extracting synchronized data at the bottom of the database</td><td>Optional</td><td>iotdb-extractor</td><td>Integrate all historical data of the database and subsequent realtime data into the sync task</td><td>no</td></tr><tr><td>processor</td><td>Pipe Processor plugin, for processing data</td><td>Optional</td><td>do-nothing-processor</td><td>no processing of incoming data</td><td><!----></td></tr><tr><td>connector</td><td>Pipe Connector plugin，for sending data</td><td><!----></td><td>-</td><td>-</td><td><!----></td></tr></tbody></table><p>In the example, the iotdb-extractor, do-nothing-processor, and iotdb-thrift-connector plugins are used to build the data synchronisation task. iotdb has other built-in data synchronisation plugins, **see the section &quot;System pre-built data synchronisation plugins&quot; <strong>. See the &quot;System Pre-installed Stream Processing Plugin&quot; section</strong>.</p><p><strong>An example of a minimalist CREATE PIPE statement is as follows:</strong></p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">CREATE</span> PIPE <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span> <span class="token comment">-- PipeId is a name that uniquely identifies the task.</span>
<span class="token keyword">WITH</span> CONNECTOR <span class="token punctuation">(</span>
  <span class="token comment">-- IoTDB data sending plugin with target IoTDB</span>
  <span class="token string">&#39;connector&#39;</span>      <span class="token operator">=</span> <span class="token string">&#39;iotdb-thrift-connector&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Data service for one of the DataNode nodes on the target IoTDB ip</span>
  <span class="token string">&#39;connector.ip&#39;</span>   <span class="token operator">=</span> <span class="token string">&#39;127.0.0.1&#39;</span><span class="token punctuation">,</span>
  <span class="token comment">-- Data service port of one of the DataNode nodes of the target IoTDB</span>
  <span class="token string">&#39;connector.port&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;6667&#39;</span><span class="token punctuation">,</span>
<span class="token punctuation">)</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><p>The expressed semantics are: synchronise the full amount of historical data and subsequent arrivals of realtime data from this database instance to the IoTDB instance with target 127.0.0.1:6667.</p><p><strong>Note:</strong></p><ul><li><p>EXTRACTOR and PROCESSOR are optional, if no configuration parameters are filled in, the system will use the corresponding default implementation.</p></li><li><p>The CONNECTOR is a mandatory configuration that needs to be declared in the CREATE PIPE statement for configuring purposes.</p></li><li><p>The CONNECTOR exhibits self-reusability. For different tasks, if their CONNECTOR possesses identical KV properties (where the value corresponds to every key), <strong>the system will ultimately create only one instance of the CONNECTOR</strong> to achieve resource reuse for connections.</p><ul><li>For example, there are the following pipe1, pipe2 task declarations:</li></ul><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">CREATE</span> PIPE pipe1
<span class="token keyword">WITH</span> CONNECTOR <span class="token punctuation">(</span>
  <span class="token string">&#39;connector&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;iotdb-thrift-connector&#39;</span><span class="token punctuation">,</span>
  <span class="token string">&#39;connector.thrift.host&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;localhost&#39;</span><span class="token punctuation">,</span>
  <span class="token string">&#39;connector.thrift.port&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;9999&#39;</span><span class="token punctuation">,</span>
<span class="token punctuation">)</span>

<span class="token keyword">CREATE</span> PIPE pipe2
<span class="token keyword">WITH</span> CONNECTOR <span class="token punctuation">(</span>
  <span class="token string">&#39;connector&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;iotdb-thrift-connector&#39;</span><span class="token punctuation">,</span>
  <span class="token string">&#39;connector.thrift.port&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;9999&#39;</span><span class="token punctuation">,</span>
  <span class="token string">&#39;connector.thrift.host&#39;</span> <span class="token operator">=</span> <span class="token string">&#39;localhost&#39;</span><span class="token punctuation">,</span>
<span class="token punctuation">)</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><ul><li>Since they have identical CONNECTOR declarations (<strong>even if the order of some properties is different</strong>), the framework will automatically reuse the CONNECTOR declared by them. Hence, the CONNECTOR instances for pipe1 and pipe2 will be the same.</li></ul></li><li><p>Please note that we should avoid constructing application scenarios that involve data cycle sync (as it can result in an infinite loop):</p><ul><li>IoTDB A -&gt; IoTDB B -&gt; IoTDB A</li><li>IoTDB A -&gt; IoTDB A</li></ul></li></ul><h3 id="start-stream-processing-task" tabindex="-1"><a class="header-anchor" href="#start-stream-processing-task"><span>Start Stream Processing Task</span></a></h3><p>After the successful execution of the CREATE PIPE statement, an instance of the stream processing task is created, but the overall task&#39;s running status will be set to STOPPED, meaning the task will not immediately process data.</p><p>You can use the START PIPE statement to make the stream processing task start processing data:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">START</span> PIPE <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><h3 id="stop-stream-processing-task" tabindex="-1"><a class="header-anchor" href="#stop-stream-processing-task"><span>Stop Stream Processing Task</span></a></h3><p>Use the STOP PIPE statement to stop the stream processing task from processing data:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code>STOP PIPE <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><h3 id="delete-stream-processing-task" tabindex="-1"><a class="header-anchor" href="#delete-stream-processing-task"><span>Delete Stream Processing Task</span></a></h3><p>If a stream processing task is in the RUNNING state, you can use the DROP PIPE statement to stop it and delete the entire task:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">DROP</span> PIPE <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><p>Before deleting a stream processing task, there is no need to execute the STOP operation.</p><h3 id="show-stream-processing-task" tabindex="-1"><a class="header-anchor" href="#show-stream-processing-task"><span>Show Stream Processing Task</span></a></h3><p>Use the SHOW PIPES statement to view all stream processing tasks:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">SHOW</span> PIPES
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><p>The query results are as follows:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token operator">+</span><span class="token comment">-----------+-----------------------+-------+-------------+-------------+-------------+----------------+</span>
<span class="token operator">|</span>         ID<span class="token operator">|</span>          CreationTime <span class="token operator">|</span>  State<span class="token operator">|</span>PipeExtractor<span class="token operator">|</span>PipeProcessor<span class="token operator">|</span>PipeConnector<span class="token operator">|</span>ExceptionMessage<span class="token operator">|</span>
<span class="token operator">+</span><span class="token comment">-----------+-----------------------+-------+-------------+-------------+-------------+----------------+</span>
<span class="token operator">|</span>iotdb<span class="token operator">-</span>kafka<span class="token operator">|</span><span class="token number">2022</span><span class="token operator">-</span><span class="token number">03</span><span class="token operator">-</span><span class="token number">30</span>T20:<span class="token number">58</span>:<span class="token number">30.689</span><span class="token operator">|</span>RUNNING<span class="token operator">|</span>          <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span>          <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span>          <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span>            None<span class="token operator">|</span>
<span class="token operator">+</span><span class="token comment">-----------+-----------------------+-------+-------------+-------------+-------------+----------------+</span>
<span class="token operator">|</span>iotdb<span class="token operator">-</span>iotdb<span class="token operator">|</span><span class="token number">2022</span><span class="token operator">-</span><span class="token number">03</span><span class="token operator">-</span><span class="token number">31</span>T12:<span class="token number">55</span>:<span class="token number">28.129</span><span class="token operator">|</span>STOPPED<span class="token operator">|</span>          <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span>          <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span>          <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span> TException: <span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token punctuation">.</span><span class="token operator">|</span>
<span class="token operator">+</span><span class="token comment">-----------+-----------------------+-------+-------------+-------------+-------------+----------------+</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div><p>You can use <code>&lt;PipeId&gt;</code> to specify the status of a stream processing task you want to see:</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">SHOW</span> PIPE <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div></div></div><p>Additionally, the WHERE clause can be used to determine if the Pipe Connector used by a specific &lt;PipeId&gt; is being reused.</p><div class="language-sql line-numbers-mode" data-ext="sql" data-title="sql"><pre class="language-sql"><code><span class="token keyword">SHOW</span> PIPES
<span class="token keyword">WHERE</span> CONNECTOR USED <span class="token keyword">BY</span> <span class="token operator">&lt;</span>PipeId<span class="token operator">&gt;</span>
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div></div></div><h3 id="stream-processing-task-running-status-migration" tabindex="-1"><a class="header-anchor" href="#stream-processing-task-running-status-migration"><span>Stream Processing Task Running Status Migration</span></a></h3><p>A stream processing task status can transition through several states during the lifecycle of a data synchronization pipe:</p><ul><li><strong>STOPPED：</strong> The pipe is in a stopped state. It can have the following possibilities: <ul><li>After the successful creation of a pipe, its initial state is set to stopped</li><li>The user manually pauses a pipe that is in normal running state, transitioning its status from RUNNING to STOPPED</li><li>If a pipe encounters an unrecoverable error during execution, its status automatically changes from RUNNING to STOPPED.</li></ul></li><li><strong>RUNNING：</strong> The pipe is actively processing data</li><li><strong>DROPPED：</strong> The pipe is permanently deleted</li></ul><p>The following diagram illustrates the different states and their transitions:</p><figure><img src="https://alioss.timecho.com/docs/img/状态迁移图.png" alt="state migration diagram" tabindex="0" loading="lazy"><figcaption>state migration diagram</figcaption></figure><h2 id="authority-management" tabindex="-1"><a class="header-anchor" href="#authority-management"><span>Authority Management</span></a></h2><h3 id="stream-processing-task" tabindex="-1"><a class="header-anchor" href="#stream-processing-task"><span>Stream Processing Task</span></a></h3><table><thead><tr><th>Authority Name</th><th>Description</th></tr></thead><tbody><tr><td>CREATE_PIPE</td><td>Register task,path-independent</td></tr><tr><td>START_PIPE</td><td>Start task,path-independent</td></tr><tr><td>STOP_PIPE</td><td>Stop task,path-independent</td></tr><tr><td>DROP_PIPE</td><td>Uninstall task,path-independent</td></tr><tr><td>SHOW_PIPES</td><td>Query task,path-independent</td></tr></tbody></table><h3 id="stream-processing-task-plugin" tabindex="-1"><a class="header-anchor" href="#stream-processing-task-plugin"><span>Stream Processing Task Plugin</span></a></h3><table><thead><tr><th>Authority Name</th><th>Description</th></tr></thead><tbody><tr><td>CREATE_PIPEPLUGIN</td><td>Register stream processing task plugin,path-independent</td></tr><tr><td>DROP_PIPEPLUGIN</td><td>Delete stream processing task plugin,path-independent</td></tr><tr><td>SHOW_PIPEPLUGINS</td><td>Query stream processing task plugin,path-independent</td></tr></tbody></table><h2 id="configure-parameters" tabindex="-1"><a class="header-anchor" href="#configure-parameters"><span>Configure Parameters</span></a></h2><p>In iotdb-common.properties ：</p><div class="language-Properties line-numbers-mode" data-ext="Properties" data-title="Properties"><pre class="language-Properties"><code>####################
### Pipe Configuration
####################

# Uncomment the following field to configure the pipe lib directory.
# For Windows platform
# If its prefix is a drive specifier followed by &quot;\\&quot;, or if its prefix is &quot;\\\\&quot;, then the path is
# absolute. Otherwise, it is relative.
# pipe_lib_dir=ext\\pipe
# For Linux platform
# If its prefix is &quot;/&quot;, then the path is absolute. Otherwise, it is relative.
# pipe_lib_dir=ext/pipe

# The maximum number of threads that can be used to execute the pipe subtasks in PipeSubtaskExecutor.
# The actual value will be min(pipe_subtask_executor_max_thread_num, max(1, CPU core number / 2)).
# pipe_subtask_executor_max_thread_num=5

# The connection timeout (in milliseconds) for the thrift client.
# pipe_connector_timeout_ms=900000
</code></pre><div class="line-numbers" aria-hidden="true"><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div><div class="line-number"></div></div></div></div><!--[--><!----><!--]--><footer class="vp-page-meta"><div class="vp-meta-item edit-link"><a href="https://github.com/apache/iotdb-docs/edit/main/src/UserGuide/V1.2.x/User-Manual/Streaming.md" rel="noopener noreferrer" target="_blank" aria-label="Found Error? Edit this page on GitHub" class="nav-link vp-meta-label"><!--[--><svg xmlns="http://www.w3.org/2000/svg" class="icon edit-icon" viewBox="0 0 1024 1024" fill="currentColor" aria-label="edit icon"><path d="M430.818 653.65a60.46 60.46 0 0 1-50.96-93.281l71.69-114.012 7.773-10.365L816.038 80.138A60.46 60.46 0 0 1 859.225 62a60.46 60.46 0 0 1 43.186 18.138l43.186 43.186a60.46 60.46 0 0 1 0 86.373L588.879 565.55l-8.637 8.637-117.466 68.234a60.46 60.46 0 0 1-31.958 11.229z"></path><path d="M728.802 962H252.891A190.883 190.883 0 0 1 62.008 771.98V296.934a190.883 190.883 0 0 1 190.883-192.61h267.754a60.46 60.46 0 0 1 0 120.92H252.891a69.962 69.962 0 0 0-69.098 69.099V771.98a69.962 69.962 0 0 0 69.098 69.098h475.911A69.962 69.962 0 0 0 797.9 771.98V503.363a60.46 60.46 0 1 1 120.922 0V771.98A190.883 190.883 0 0 1 728.802 962z"></path></svg><!--]-->Found Error? Edit this page on GitHub<span><svg class="external-link-icon" xmlns="http://www.w3.org/2000/svg" aria-hidden="true" focusable="false" x="0px" y="0px" viewBox="0 0 100 100" width="15" height="15"><path fill="currentColor" d="M18.8,85.1h56l0,0c2.2,0,4-1.8,4-4v-32h-8v28h-48v-48h28v-8h-32l0,0c-2.2,0-4,1.8-4,4v56C14.8,83.3,16.6,85.1,18.8,85.1z"></path><polygon fill="currentColor" points="45.7,48.7 51.3,54.3 77.2,28.5 77.2,37.2 85.2,37.2 85.2,14.9 62.8,14.9 62.8,22.9 71.5,22.9"></polygon></svg><span class="external-link-icon-sr-only">open in new window</span></span><!----></a></div><div class="vp-meta-item git-info"><div class="update-time"><span class="vp-meta-label">Last update: </span><!----></div><!----></div></footer><nav class="vp-page-nav"><a class="route-link nav-link prev" href="/UserGuide/V1.2.x/User-Manual/Operator-and-Expression.html" aria-label="Operator and Expression"><div class="hint"><span class="arrow start"></span>Prev</div><div class="link"><!---->Operator and Expression</div></a><a class="route-link nav-link next" href="/UserGuide/V1.2.x/User-Manual/Data-Sync.html" aria-label="Data Sync"><div class="hint">Next<span class="arrow end"></span></div><div class="link">Data Sync<!----></div></a></nav><!----><!--[--><!----><!--]--><!--]--></main><!--]--><footer style="padding-bottom:2rem;"><span id="doc-version" style="display:none;">rel/1.2</span><p style="text-align:center;color:#909399;font-size:12px;margin:0 30px;">Copyright © 2024 The Apache Software Foundation.<br> Apache and the Apache feather logo are trademarks of The Apache Software Foundation</p><p style="text-align:center;margin-top:10px;color:#909399;font-size:12px;margin:0 30px;"><strong>Have a question?</strong> Connect with us on QQ, WeChat, or Slack. <a href="https://github.com/apache/iotdb/issues/1995">Join the community</a> now.</p></footer></div><!--]--><!--]--><!--[--><!----><!--]--><!--]--></div>
    <script type="module" src="/assets/app-0jI7BR7G.js" defer></script>
  </body>
</html>