| <!doctype html><html lang=en class=no-js><head><meta charset=utf-8><meta http-equiv=x-ua-compatible content="IE=edge"><meta name=viewport content="width=device-width,initial-scale=1"><title>Euphoria Java 8 DSL</title><meta name=description content="Apache Beam is an open source, unified model and set of language-specific SDKs for defining and executing data processing workflows, and also data ingestion and integration flows, supporting Enterprise Integration Patterns (EIPs) and Domain Specific Languages (DSLs). Dataflow pipelines simplify the mechanics of large-scale batch and streaming data processing and can run on a number of runtimes like Apache Flink, Apache Spark, and Google Cloud Dataflow (a cloud service). Beam also brings DSL in different languages, allowing users to easily implement their data integration processes."><link href="https://fonts.googleapis.com/css?family=Roboto:100,300,400,500,700" rel=stylesheet><link rel=preload href=/scss/main.min.408fddfe3e8a45f87a5a8c9a839d77db667c1c534e5e5cd0d957ffc3dd6c14cf.css as=style><link href=/scss/main.min.408fddfe3e8a45f87a5a8c9a839d77db667c1c534e5e5cd0d957ffc3dd6c14cf.css rel=stylesheet integrity><script src=https://code.jquery.com/jquery-2.2.4.min.js></script><style>.body__contained img{max-width:100%}</style><script type=text/javascript src=/js/bootstrap.min.2979f9a6e32fc42c3e7406339ee9fe76b31d1b52059776a02b4a7fa6a4fd280a.js defer></script> |
| <script type=text/javascript src=/js/language-switch-v2.min.121952b7980b920320ab229551857669209945e39b05ba2b433a565385ca44c6.js defer></script> |
| <script type=text/javascript src=/js/fix-menu.min.039174b67107465f2090a493f91e126f7aa797f29420f9edab8a54d9dd4b3d2d.js defer></script> |
| <script type=text/javascript src=/js/section-nav.min.1405fd5e70fab5f6c54037c269b1d137487d8f3d1b3009032525f6db3fbce991.js defer></script> |
| <script type=text/javascript src=/js/page-nav.min.af231204c9c52c5089d53a4c02739eacbb7f939e3be1c6ffcc212e0ac4dbf879.js defer></script> |
| <script type=text/javascript src=/js/expandable-list.min.75a4526624a3b8898fe7fb9e3428c205b581f8b38c7926922467aef17eac69f2.js defer></script> |
| <script type=text/javascript src=/js/copy-to-clipboard.min.364c06423d7e8993fc42bb4abc38c03195bc8386db26d18774ce775d08d5b18d.js defer></script> |
| <script type=text/javascript src=/js/calendar.min.336664054fa0f52b08bbd4e3c59b5cb6d63dcfb2b4d602839746516b0817446b.js defer></script> |
| <script type=text/javascript src=/js/fix-playground-nested-scroll.min.0283f1037cb1b9d5074c6eaf041292b524a8148a7cdb803d5ccd6d1fc4eb3253.js defer></script> |
| <script type=text/javascript src=/js/anchor-content-jump-fix.min.22d3240f81632e4c11179b9d2aaf37a40da9414333c43aa97344e8b21a7df0e4.js defer></script> |
| <link rel=alternate type=application/rss+xml title="Apache Beam" href=/feed.xml><link rel=canonical href=/documentation/sdks/java/euphoria/ data-proofer-ignore><link rel="shortcut icon" type=image/x-icon href=/images/favicon.ico><link rel=stylesheet href=https://use.fontawesome.com/releases/v5.4.1/css/all.css integrity=sha384-5sAR7xN1Nv6T6+dT2mhtzEpVJvfS3NScPQTrOxhwjIuvcA67KV2R5Jz6kr4abQsz crossorigin=anonymous><link rel=stylesheet href=https://unpkg.com/swiper@8/swiper-bundle.min.css><script async src=https://platform.twitter.com/widgets.js></script> |
| <script>(function(e,t,n,s,o,i,a){e.GoogleAnalyticsObject=o,e[o]=e[o]||function(){(e[o].q=e[o].q||[]).push(arguments)},e[o].l=1*new Date,i=t.createElement(n),a=t.getElementsByTagName(n)[0],i.async=1,i.src=s,a.parentNode.insertBefore(i,a)})(window,document,"script","//www.google-analytics.com/analytics.js","ga"),ga("create","UA-73650088-1","auto"),ga("send","pageview")</script><script>(function(e,t,n,s,o,i){e.hj=e.hj||function(){(e.hj.q=e.hj.q||[]).push(arguments)},e._hjSettings={hjid:2182187,hjsv:6},o=t.getElementsByTagName("head")[0],i=t.createElement("script"),i.async=1,i.src=n+e._hjSettings.hjid+s+e._hjSettings.hjsv,o.appendChild(i)})(window,document,"https://static.hotjar.com/c/hotjar-",".js?sv=")</script></head><body class=body data-spy=scroll data-target=.page-nav data-offset=0><nav class="navigation-bar-mobile header navbar navbar-fixed-top"><div class=navbar-header><a href=/ class=navbar-brand><img alt=Brand style=height:46px;width:43px src=/images/beam_logo_navbar_mobile.png></a> |
| <a class=navbar-link href=/get-started/>Get Started</a> |
| <a class=navbar-link href=/documentation/>Documentation</a> |
| <button type=button class="navbar-toggle menu-open" aria-expanded=false aria-controls=navbar onclick=openMenu()> |
| <span class=sr-only>Toggle navigation</span> |
| <span class=icon-bar></span> |
| <span class=icon-bar></span> |
| <span class=icon-bar></span></button></div><div class="navbar-mask closed"></div><div id=navbar class="navbar-container closed"><button type=button class=navbar-toggle aria-expanded=false aria-controls=navbar id=closeMenu> |
| <span class=sr-only>Toggle navigation</span> |
| <span class=icon-bar></span> |
| <span class=icon-bar></span> |
| <span class=icon-bar></span></button><ul class="nav navbar-nav"><li><div class=searchBar-mobile><script>(function(){var t,n="012923275103528129024:4emlchv9wzi",e=document.createElement("script");e.type="text/javascript",e.async=!0,e.src="https://cse.google.com/cse.js?cx="+n,t=document.getElementsByTagName("script")[0],t.parentNode.insertBefore(e,t)})()</script><gcse:search></gcse:search></div></li><li><a class=navbar-link href=/about>About</a></li><li><a class=navbar-link href=/get-started/>Get Started</a></li><li><span class=navbar-link>Documentation</span><ul><li><a href=/documentation/>General</a></li><li><a href=/documentation/sdks/java/>Languages</a></li><li><a href=/documentation/runners/capability-matrix/>Runners</a></li><li><a href=/documentation/io/connectors/>I/O Connectors</a></li></ul></li><li><a class=navbar-link href=/roadmap/>Roadmap</a></li><li><a class=navbar-link href=/community/>Community</a></li><li><a class=navbar-link href=/contribute/>Contribute</a></li><li><a class=navbar-link href=/blog/>Blog</a></li><li><a class=navbar-link href=/case-studies/>Case Studies</a></li></ul><ul class="nav navbar-nav navbar-right"><li><a href=https://github.com/apache/beam/edit/master/website/www/site/content/en/documentation/sdks/java/euphoria.md data-proofer-ignore><svg xmlns="http://www.w3.org/2000/svg" width="25" height="24" fill="none" viewBox="0 0 25 24"><path stroke="#ff6d00" stroke-linecap="round" stroke-linejoin="round" stroke-width="2.75" d="M4.543 20h4l10.5-10.5c.53-.53.828-1.25.828-2s-.298-1.47-.828-2-1.25-.828-2-.828-1.47.298-2 .828L4.543 16v4zm9.5-13.5 4 4"/></svg></a></li><li class=dropdown><a href=# class=dropdown-toggle id=apache-dropdown data-toggle=dropdown role=button aria-haspopup=true aria-expanded=false><img src=https://www.apache.org/foundation/press/kit/feather_small.png alt="Apache Logo" style=height:20px> |
| Apache |
| <span class=arrow-icon><svg xmlns="http://www.w3.org/2000/svg" width="20" height="20" fill="none" viewBox="0 0 20 20"><circle cx="10" cy="10" r="10" fill="#ff6d00"/><path stroke="#fff" stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M8.535 5.28l4.573 4.818-4.573 4.403"/></svg></span></a><ul class="dropdown-menu dropdown-menu-right"><li><a target=_blank href=https://www.apache.org/>ASF Homepage</a></li><li><a target=_blank href=https://www.apache.org/licenses/>License</a></li><li><a target=_blank href=https://www.apache.org/security/>Security</a></li><li><a target=_blank href=https://www.apache.org/foundation/thanks.html>Thanks</a></li><li><a target=_blank href=https://www.apache.org/foundation/sponsorship.html>Sponsorship</a></li><li><a target=_blank href=https://www.apache.org/foundation/policies/conduct>Code of Conduct</a></li></ul></li></ul></div></nav><nav class=navigation-bar-desktop><a href=/ class=navbar-logo><img src=/images/beam_logo_navbar.png alt="Beam Logo"></a><div class=navbar-bar-left><div class=navbar-links><a class=navbar-link href=/about>About</a> |
| <a class=navbar-link href=/get-started/>Get Started</a><li class="dropdown navbar-dropdown navbar-dropdown-documentation"><a href=# class="dropdown-toggle navbar-link" role=button aria-haspopup=true aria-expanded=false>Documentation |
| <span><svg xmlns="http://www.w3.org/2000/svg" width="12" height="11" fill="none" viewBox="0 0 12 11"><path stroke="#ff6d00" stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M10.666 4.535 5.847 9.108 1.444 4.535"/></svg></span></a><ul class=dropdown-menu><li><a class=navbar-dropdown-menu-link href=/documentation/>General</a></li><li><a class=navbar-dropdown-menu-link href=/documentation/sdks/java/>Languages</a></li><li><a class=navbar-dropdown-menu-link href=/documentation/runners/capability-matrix/>Runners</a></li><li><a class=navbar-dropdown-menu-link href=/documentation/io/connectors/>I/O Connectors</a></li></ul></li><a class=navbar-link href=/roadmap/>Roadmap</a> |
| <a class=navbar-link href=/community/>Community</a> |
| <a class=navbar-link href=/contribute/>Contribute</a> |
| <a class=navbar-link href=/blog/>Blog</a> |
| <a class=navbar-link href=/case-studies/>Case Studies</a></div><div id=iconsBar><a type=button onclick=showSearch()><svg xmlns="http://www.w3.org/2000/svg" width="25" height="24" fill="none" viewBox="0 0 25 24"><path stroke="#ff6d00" stroke-linecap="round" stroke-linejoin="round" stroke-width="2.75" d="M10.191 17c3.866.0 7-3.134 7-7s-3.134-7-7-7-7 3.134-7 7 3.134 7 7 7zm11 4-6-6"/></svg></a><a target=_blank href=https://github.com/apache/beam/edit/master/website/www/site/content/en/documentation/sdks/java/euphoria.md data-proofer-ignore><svg xmlns="http://www.w3.org/2000/svg" width="25" height="24" fill="none" viewBox="0 0 25 24"><path stroke="#ff6d00" stroke-linecap="round" stroke-linejoin="round" stroke-width="2.75" d="M4.543 20h4l10.5-10.5c.53-.53.828-1.25.828-2s-.298-1.47-.828-2-1.25-.828-2-.828-1.47.298-2 .828L4.543 16v4zm9.5-13.5 4 4"/></svg></a><li class="dropdown navbar-dropdown navbar-dropdown-apache"><a href=# class=dropdown-toggle role=button aria-haspopup=true aria-expanded=false><img src=https://www.apache.org/foundation/press/kit/feather_small.png alt="Apache Logo" style=height:20px> |
| Apache |
| <span class=arrow-icon><svg xmlns="http://www.w3.org/2000/svg" width="20" height="20" fill="none" viewBox="0 0 20 20"><circle cx="10" cy="10" r="10" fill="#ff6d00"/><path stroke="#fff" stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M8.535 5.28l4.573 4.818-4.573 4.403"/></svg></span></a><ul class=dropdown-menu><li><a class=navbar-dropdown-menu-link target=_blank href=https://www.apache.org/>ASF Homepage</a></li><li><a class=navbar-dropdown-menu-link target=_blank href=https://www.apache.org/licenses/>License</a></li><li><a class=navbar-dropdown-menu-link target=_blank href=https://www.apache.org/security/>Security</a></li><li><a class=navbar-dropdown-menu-link target=_blank href=https://www.apache.org/foundation/thanks.html>Thanks</a></li><li><a class=navbar-dropdown-menu-link target=_blank href=https://www.apache.org/foundation/sponsorship.html>Sponsorship</a></li><li><a class=navbar-dropdown-menu-link target=_blank href=https://www.apache.org/foundation/policies/conduct>Code of Conduct</a></li></ul></li></div><div class="searchBar disappear"><script>(function(){var t,n="012923275103528129024:4emlchv9wzi",e=document.createElement("script");e.type="text/javascript",e.async=!0,e.src="https://cse.google.com/cse.js?cx="+n,t=document.getElementsByTagName("script")[0],t.parentNode.insertBefore(e,t)})()</script><gcse:search></gcse:search> |
| <a type=button onclick=endSearch()><svg xmlns="http://www.w3.org/2000/svg" width="25" height="25" fill="none" viewBox="0 0 25 25"><path stroke="#ff6d00" stroke-linecap="round" stroke-linejoin="round" stroke-width="2.75" d="M21.122 20.827 4.727 4.432M21.122 4.43 4.727 20.827"/></svg></a></div></div></nav><div class=header-push></div><div class="top-banners swiper"><div class=swiper-wrapper><div class=swiper-slide><a href=https://tour.beam.apache.org><img class=banner-img-desktop src=/images/banners/tour-of-beam/tour-of-beam-desktop.png alt="Start Tour of Beam"> |
| <img class=banner-img-mobile src=/images/banners/tour-of-beam/tour-of-beam-mobile.png alt="Start Tour of Beam"></a></div><div class=swiper-slide><a href=https://beam.apache.org/documentation/ml/overview/><img class=banner-img-desktop src=/images/banners/machine-learning/machine-learning-desktop.jpg alt="Machine Learning"> |
| <img class=banner-img-mobile src=/images/banners/machine-learning/machine-learning-mobile.jpg alt="Machine Learning"></a></div></div><div class=swiper-pagination></div><div class=swiper-button-prev></div><div class=swiper-button-next></div></div><script src=/js/swiper-bundle.min.min.e0e8f81b0b15728d35ff73c07f42ddbb17a108d6f23df4953cb3e60df7ade675.js></script> |
| <script src=/js/sliders/top-banners.min.afa7d0a19acf7a3b28ca369490b3d401a619562a2a4c9612577be2f66a4b9855.js></script> |
| <script>function showSearch(){addPlaceholder();var e,t=document.querySelector(".searchBar");t.classList.remove("disappear"),e=document.querySelector("#iconsBar"),e.classList.add("disappear")}function addPlaceholder(){$("input:text").attr("placeholder","What are you looking for?")}function endSearch(){var e,t=document.querySelector(".searchBar");t.classList.add("disappear"),e=document.querySelector("#iconsBar"),e.classList.remove("disappear")}function blockScroll(){$("body").toggleClass("fixedPosition")}function openMenu(){addPlaceholder(),blockScroll()}</script><div class="clearfix container-main-content"><div class="section-nav closed" data-offset-top=90 data-offset-bottom=500><span class="section-nav-back glyphicon glyphicon-menu-left"></span><nav><ul class=section-nav-list data-section-nav><li><span class=section-nav-list-main-title>Languages</span></li><li><span class=section-nav-list-title>Java</span><ul class=section-nav-list><li><a href=/documentation/sdks/java/>Java SDK overview</a></li><li><a href=https://beam.apache.org/releases/javadoc/2.55.1/ target=_blank>Java SDK API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></li><li><a href=/documentation/sdks/java-dependencies/>Java SDK dependencies</a></li><li><a href=/documentation/sdks/java-extensions/>Java SDK extensions</a></li><li><a href=/documentation/sdks/java-thirdparty/>Java 3rd party extensions</a></li><li><a href=/documentation/sdks/java/testing/nexmark/>Nexmark benchmark suite</a></li><li><a href=/documentation/sdks/java/testing/tpcds/>TPC-DS benchmark suite</a></li><li><a href=/documentation/sdks/java-multi-language-pipelines/>Java multi-language pipelines quickstart</a></li></ul></li><li><span class=section-nav-list-title>Python</span><ul class=section-nav-list><li><a href=/documentation/sdks/python/>Python SDK overview</a></li><li><a href=https://beam.apache.org/releases/pydoc/2.55.1/ target=_blank>Python SDK API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></li><li><a href=/documentation/sdks/python-dependencies/>Python SDK dependencies</a></li><li><a href=/documentation/sdks/python-streaming/>Python streaming pipelines</a></li><li><a href=/documentation/sdks/python-type-safety/>Ensuring Python type safety</a></li><li><a href=/documentation/sdks/python-machine-learning/>Machine Learning</a></li><li><a href=/documentation/sdks/python-pipeline-dependencies/>Managing pipeline dependencies</a></li><li><a href=/documentation/sdks/python-multi-language-pipelines/>Python multi-language pipelines quickstart</a></li><li><a href=/documentation/sdks/python-unrecoverable-errors/>Python Unrecoverable Errors</a></li></ul></li><li><span class=section-nav-list-title>Go</span><ul class=section-nav-list><li><a href=/documentation/sdks/go/>Go SDK overview</a></li><li><a href=https://pkg.go.dev/github.com/apache/beam/sdks/v2/go/pkg/beam target=_blank>Go SDK API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a><li><a href=/documentation/sdks/go-dependencies/>Go SDK dependencies</a></li><li><a href=/documentation/sdks/go-cross-compilation/>Cross compilation</a></li></li></ul></li><li><span class=section-nav-list-title>Typescript</span><ul class=section-nav-list><li><a href=/documentation/sdks/typescript/>Typescript SDK overview</a></li><li><a href=https://beam.apache.org/releases/typedoc/current/ target=_blank>Typescript SDK API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></li></ul></li><li><span class=section-nav-list-title>Scala</span><ul class=section-nav-list><li><a href=/documentation/sdks/scala/>Scio</a></li><li><a href=https://spotify.github.io/scio/api/com/spotify/scio/index.html target=_blank>Scio SDK API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></li></ul></li><li><span class=section-nav-list-title>Yaml</span><ul class=section-nav-list><li><a href=/documentation/sdks/yaml/>Yaml overview</a></li><li><a href=/documentation/sdks/yaml-udf/>Yaml User Defined Functions</a></li><li><a href=/documentation/sdks/yaml-combine/>Yaml Aggregation</a></li><li><a href=/documentation/sdks/yaml-errors/>Error handling</a></li><li><a href=/documentation/sdks/yaml-inline-python/>Inlining Python</a></li><li><a href=https://beam.apache.org/releases/yamldoc/current/ target=_blank>YAML API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></ul></li><li><span class=section-nav-list-title>SQL</span><ul class=section-nav-list><li><a href=/documentation/dsls/sql/overview/>Overview</a></li><li><a href=/documentation/dsls/sql/walkthrough/>Walkthrough</a></li><li><a href=/documentation/dsls/sql/shell/>Shell</a></li><li class=section-nav-item--collapsible><span class=section-nav-list-title>Apache Calcite dialect</span><ul class=section-nav-list><li><a href=/documentation/dsls/sql/calcite/overview/>Calcite support overview</a></li><li><a href=/documentation/dsls/sql/calcite/query-syntax/>Query syntax</a></li><li><a href=/documentation/dsls/sql/calcite/lexical/>Lexical structure</a></li><li><a href=/documentation/dsls/sql/calcite/data-types/>Data types</a></li><li><a href=/documentation/dsls/sql/calcite/scalar-functions/>Scalar functions</a></li><li><a href=/documentation/dsls/sql/calcite/aggregate-functions/>Aggregate functions</a></li></ul></li><li class=section-nav-item--collapsible><span class=section-nav-list-title>ZetaSQL dialect</span><ul class=section-nav-list><li><a href=/documentation/dsls/sql/zetasql/overview/>ZetaSQL support overview</a></li><li><a href=/documentation/dsls/sql/zetasql/syntax/>Function call rules</a></li><li><a href=/documentation/dsls/sql/zetasql/conversion-rules/>Conversion rules</a></li><li><a href=/documentation/dsls/sql/zetasql/query-syntax/>Query syntax</a></li><li><a href=/documentation/dsls/sql/zetasql/lexical/>Lexical structure</a></li><li><a href=/documentation/dsls/sql/zetasql/data-types/>Data types</a></li><li><a href=/documentation/dsls/sql/zetasql/operators/>Operators</a></li><li class=section-nav-item--collapsible><span class=section-nav-list-title>Scalar functions</span><ul class=section-nav-list><li><a href=/documentation/dsls/sql/zetasql/string-functions/>String functions</a></li><li><a href=/documentation/dsls/sql/zetasql/math-functions/>Mathematical functions</a></li><li><a href=/documentation/dsls/sql/zetasql/conditional-expressions/>Conditional expressions</a></li></ul></li><li><a href=/documentation/dsls/sql/zetasql/aggregate-functions/>Aggregate functions</a></li></ul></li><li class=section-nav-item--collapsible><span class=section-nav-list-title>Beam SQL extensions</span><ul class=section-nav-list><li><a href=/documentation/dsls/sql/extensions/create-external-table/>CREATE EXTERNAL TABLE</a></li><li><a href=/documentation/dsls/sql/extensions/windowing-and-triggering/>Windowing & triggering</a></li><li><a href=/documentation/dsls/sql/extensions/joins/>Joins</a></li><li><a href=/documentation/dsls/sql/extensions/user-defined-functions/>User-defined functions</a></li><li><a href=/documentation/dsls/sql/extensions/set/>SET pipeline options</a></li></ul></li></ul></li><li><span class=section-nav-list-title>DataFrames</span><ul class=section-nav-list><li><a href=/documentation/dsls/dataframes/overview/>Overview</a></li><li><a href=/documentation/dsls/dataframes/differences-from-pandas/>Differences from pandas</a></li><li><a href=https://github.com/apache/beam/tree/master/sdks/python/apache_beam/examples/dataframe target=_blank>Example pipelines <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></li><li><a href=https://beam.apache.org/releases/pydoc/2.55.1/apache_beam.dataframe.html target=_blank>DataFrame API reference <img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></li></ul></li></ul></nav></div><nav class="page-nav clearfix" data-offset-top=90 data-offset-bottom=500><nav id=TableOfContents><ul><li><a href=#what-is-euphoria>What is Euphoria</a></li><li><a href=#wordcount-example>WordCount Example</a></li><li><a href=#euphoria-guide>Euphoria Guide</a><ul><li><a href=#inputs-and-outputs>Inputs and Outputs</a></li><li><a href=#adding-operators>Adding Operators</a></li><li><a href=#coders-and-types>Coders and Types</a></li><li><a href=#metrics-and-accumulators>Metrics and Accumulators</a></li><li><a href=#windowing>Windowing</a></li></ul></li><li><a href=#how-to-get-euphoria>How to get Euphoria</a></li><li><a href=#operator-reference>Operator Reference</a><ul><li><a href=#countbykey><code>CountByKey</code></a></li><li><a href=#distinct><code>Distinct</code></a></li><li><a href=#join><code>Join</code></a></li><li><a href=#leftjoin><code>LeftJoin</code></a></li><li><a href=#rightjoin><code>RightJoin</code></a></li><li><a href=#fulljoin><code>FullJoin</code></a></li><li><a href=#mapelements><code>MapElements</code></a></li><li><a href=#flatmap><code>FlatMap</code></a></li><li><a href=#filter><code>Filter</code></a></li><li><a href=#reducebykey><code>ReduceByKey</code></a></li><li><a href=#reducewindow><code>ReduceWindow</code></a></li><li><a href=#sumbykey><code>SumByKey</code></a></li><li><a href=#union><code>Union</code></a></li><li><a href=#topperkey><code>TopPerKey</code></a></li><li><a href=#assigneventtime><code>AssignEventTime</code></a></li></ul></li><li><a href=#translation>Translation</a><ul><li><a href=#translationproviders>TranslationProviders</a><ul><li><a href=#generictranslatorprovider><code>GenericTranslatorProvider</code></a></li><li><a href=#compositeprovider><code>CompositeProvider</code></a></li></ul></li><li><a href=#operator-translators>Operator Translators</a><ul><li><a href=#broadcasthashjointranslator><code>BroadcastHashJoinTranslator</code></a></li><li><a href=#compositeoperatortranslator><code>CompositeOperatorTranslator</code></a></li></ul></li><li><a href=#details>Details</a></li></ul></li><li><a href=#unsupported-features>Unsupported Features</a></li></ul></nav></nav><div class="body__contained body__section-nav"><h1 id=euphoria-java-8-dsl>Euphoria Java 8 DSL</h1><h2 id=what-is-euphoria>What is Euphoria</h2><p>Easy to use Java 8 API build on top of the Beam’s Java SDK. API provides a <a href=#operator-reference>high-level abstraction</a> of data transformations, with focus on the Java 8 language features (e.g. lambdas and streams). It is fully inter-operable with existing Beam SDK and convertible back and forth. It allows fast prototyping through use of (optional) <a href=https://github.com/EsotericSoftware/kryo>Kryo</a> based coders, lambdas and high level operators and can be seamlessly integrated into existing Beam <code>Pipelines</code>.</p><p><a href=https://github.com/seznam/euphoria>Euphoria API</a> project has been started in 2014, with a clear goal of providing the main building block for <a href=https://www.seznam.cz/>Seznam.cz’s</a> data infrastructure. |
| In 2015, <a href=https://www.vldb.org/pvldb/vol8/p1792-Akidau.pdf>DataFlow whitepaper</a> inspired original authors to go one step further and also provide the unified API for both stream and batch processing. |
| The API has been open-sourced in 2016 and is still in active development. As the Beam’s community goal was very similar, we decided to contribute |
| the API as a high level DSL over Beam Java SDK and share our effort with the community.</p><p>Euphoria DSL integration is still work in progress and is tracked as part of <a href=https://issues.apache.org/jira/browse/BEAM-3900>BEAM-3900</a>.</p><h2 id=wordcount-example>WordCount Example</h2><p>Lets start with the small example.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>PipelineOptions</span> <span class=n>options</span> <span class=o>=</span> <span class=n>PipelineOptionsFactory</span><span class=o>.</span><span class=na>create</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=n>Pipeline</span> <span class=n>pipeline</span> <span class=o>=</span> <span class=n>Pipeline</span><span class=o>.</span><span class=na>create</span><span class=o>(</span><span class=n>options</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=c1>// Use Kryo as coder fallback |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>KryoCoderProvider</span><span class=o>.</span><span class=na>of</span><span class=o>().</span><span class=na>registerTo</span><span class=o>(</span><span class=n>pipeline</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=c1>// Source of data loaded from Beam IO. |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>input</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>pipeline</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>apply</span><span class=o>(</span><span class=n>Create</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>textLineByLine</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>setTypeDescriptor</span><span class=o>(</span><span class=n>TypeDescriptor</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>String</span><span class=o>.</span><span class=na>class</span><span class=o>));</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=c1>// zero, one, or more output elements. From input lines we will get data set of words. |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>words</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>FlatMap</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"TOKENIZER"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>lines</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>String</span> <span class=n>line</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>context</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=k>for</span> <span class=o>(</span><span class=n>String</span> <span class=n>word</span> <span class=o>:</span> <span class=n>Splitter</span><span class=o>.</span><span class=na>onPattern</span><span class=o>(</span><span class=s>"\\s+"</span><span class=o>).</span><span class=na>split</span><span class=o>(</span><span class=n>line</span><span class=o>))</span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=n>context</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>word</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> <span class=o>}</span> |
| </span></span><span class=line><span class=cl> <span class=o>})</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=c1>// Now we can count input words - the operator ensures that all values for the same |
| </span></span></span><span class=line><span class=cl><span class=c1>// key (word in this case) end up being processed together. Then it counts number of appearances |
| </span></span></span><span class=line><span class=cl><span class=c1>// of the same key in 'words' PCollection and emits it to output. |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>String</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>counted</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>CountByKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"COUNT"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>words</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>w</span> <span class=o>-></span> <span class=n>w</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=c1>// Format output. |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>output</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>MapElements</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"FORMAT"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>counted</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span><span class=n>p</span> <span class=o>-></span> <span class=n>p</span><span class=o>.</span><span class=na>getKey</span><span class=o>()</span> <span class=o>+</span> <span class=s>": "</span> <span class=o>+</span> <span class=n>p</span><span class=o>.</span><span class=na>getValue</span><span class=o>())</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=c1>// Now we can again use Beam transformation. In this case we save words and their count |
| </span></span></span><span class=line><span class=cl><span class=c1>// into the text file. |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>output</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>apply</span><span class=o>(</span><span class=n>TextIO</span><span class=o>.</span><span class=na>write</span><span class=o>()</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>to</span><span class=o>(</span><span class=s>"counted_words"</span><span class=o>));</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>pipeline</span><span class=o>.</span><span class=na>run</span><span class=o>();</span></span></span></code></pre></div></div></div></p><h2 id=euphoria-guide>Euphoria Guide</h2><p>Euphoria API is composed from a set of operators, which allows you to construct <code>Pipeline</code> according to your application needs.</p><h3 id=inputs-and-outputs>Inputs and Outputs</h3><p>Input data can be supplied through Beams IO into <code>PCollection</code>, the same way as in Beam.</p><div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>input</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>pipeline</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>apply</span><span class=o>(</span><span class=n>Create</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=s>"mouse"</span><span class=o>,</span> <span class=s>"rat"</span><span class=o>,</span> <span class=s>"elephant"</span><span class=o>,</span> <span class=s>"cat"</span><span class=o>,</span> <span class=s>"X"</span><span class=o>,</span> <span class=s>"duck"</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>setTypeDescriptor</span><span class=o>(</span><span class=n>TypeDescriptor</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>String</span><span class=o>.</span><span class=na>class</span><span class=o>));</span></span></span></code></pre></div></div></div><h3 id=adding-operators>Adding Operators</h3><p>Real power of Euphoria API is in its <a href=#operator-reference>operators suite</a>. Each Operator consumes one or more input and produces one output |
| <code>PCollection</code>. Lets take a look at simple <code>MapElements</code> example.</p><p><div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>input</span> <span class=o>=</span> <span class=o>...</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>mappedElements</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>MapElements</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"Int2Str"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>valueOf</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span></span></span></code></pre></div></div></div>The operator consumes <code>input</code>, it applies given lambda expression (<code>String::valueOf</code>) on each element of <code>input</code> and returns mapped <code>PCollection</code>. Developer is guided through series of steps when creating operator so the declaration of an operator is straightforward. To start building operator just wrote its name and ‘.’ (dot). Your IDE will give you hints.</p><p>First step to build any operator is to give it a name through <code>named()</code> method. The name is propagated through system and can latter be used when debugging.</p><h3 id=coders-and-types>Coders and Types</h3><p>Beam’s Java SDK requires developers to supply <code>Coder</code> for custom element type in order to have a way of materializing elements. Euphoria allows to use <a href=https://github.com/EsotericSoftware/kryo>Kryo</a> as a way of serialization. The <a href=https://github.com/EsotericSoftware/kryo>Kryo</a> is located in <code>:sdks:java:extensions:kryo</code> module.</p><pre tabindex=0><code>//gradle |
| dependencies { |
| compile "org.apache.beam:sdks:java:extensions:kryo:${beam.version}" |
| } |
| </code></pre><pre tabindex=0><code>//maven |
| <dependency> |
| <groupId>org.apache.beam</groupId> |
| <artifactId>beam-sdks-java-extensions-kryo</artifactId> |
| <version>${beam.version}</version> |
| </dependency> |
| </code></pre><p>All you need is to create <code>KryoCoderProvider</code> and register it to your |
| <code>Pipeline</code>. There are two ways of doing that.</p><p>When prototyping you may decide not to care much about coders, then create <code>KryoCoderProvider</code> without any class registrations to <a href=https://github.com/EsotericSoftware/kryo>Kryo</a>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//Register `KryoCoderProvider` which attempt to use `KryoCoder` to every non-primitive type |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>KryoCoderProvider</span><span class=o>.</span><span class=na>of</span><span class=o>().</span><span class=na>registerTo</span><span class=o>(</span><span class=n>pipeline</span><span class=o>);</span></span></span></code></pre></div></div></div>Such a <code>KryoCoderProvider</code> will return <code>KryoCoder</code> for every non-primitive element type. That of course degrades performance, since Kryo is not able to serialize instance of unknown types effectively. But it boost speed of pipeline development. This behavior is enabled by default and can be disabled when creating <code>Pipeline</code> through <code>KryoOptions</code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>PipelineOptions</span> <span class=n>options</span> <span class=o>=</span> <span class=n>PipelineOptionsFactory</span><span class=o>.</span><span class=na>create</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=n>options</span><span class=o>.</span><span class=na>as</span><span class=o>(</span><span class=n>KryoOptions</span><span class=o>.</span><span class=na>class</span><span class=o>).</span><span class=na>setKryoRegistrationRequired</span><span class=o>(</span><span class=kc>true</span><span class=o>);</span></span></span></code></pre></div></div></div></p><p>Second more performance friendly way is to register all the types which will Kryo serialize. Sometimes it is also a good idea to register Kryo serializers of its own too. Euphoria allows you to do that by implementing your own <code>KryoRegistrar</code> and using it when creating <code>KryoCoderProvider</code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//Do not allow `KryoCoderProvider` to return `KryoCoder` for unregistered types |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>options</span><span class=o>.</span><span class=na>as</span><span class=o>(</span><span class=n>KryoOptions</span><span class=o>.</span><span class=na>class</span><span class=o>).</span><span class=na>setKryoRegistrationRequired</span><span class=o>(</span><span class=kc>true</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>KryoCoderProvider</span><span class=o>.</span><span class=na>of</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>kryo</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> <span class=c1>//KryoRegistrar of your uwn |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=n>kryo</span><span class=o>.</span><span class=na>register</span><span class=o>(</span><span class=n>KryoSerializedElementType</span><span class=o>.</span><span class=na>class</span><span class=o>);</span> <span class=c1>//other may follow |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>})</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>registerTo</span><span class=o>(</span><span class=n>pipeline</span><span class=o>);</span></span></span></code></pre></div></div></div>Beam resolves coders using types of elements. Type information is not available at runtime when element type is described by lambda implementation. It is due to type erasure and dynamic nature of lambda expressions. So there is an optional way of supplying <code>TypeDescriptor</code> every time new type is introduced during Operator construction.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>input</span> <span class=o>=</span> <span class=o>...</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>MapElements</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"Int2Str"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>valueOf</span><span class=o>,</span> <span class=n>TypeDescriptors</span><span class=o>.</span><span class=na>strings</span><span class=o>())</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span></span></span></code></pre></div></div></div>Euphoria operator’s will use <code>TypeDescriptor<Object></code>, when <code>TypeDescriptors</code> is not supplied by user. So <code>KryoCoderProvider</code> may return <code>KryoCoder<Object></code> for every element with unknown type, if allowed by <code>KryoOptions</code>. Supplying <code>TypeDescriptors</code> becomes mandatory when using <code>.setKryoRegistrationRequired(true)</code>.</p><h3 id=metrics-and-accumulators>Metrics and Accumulators</h3><p>Statistics about job’s internals are very helpful during development of distributed jobs. Euphoria calls them accumulators. They are accessible through environment <code>Context</code>, which can be obtained from <code>Collector</code>, whenever working with it. It is usually present when zero-to-many output elements are expected from operator. For example in case of <code>FlatMap</code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>Pipeline</span> <span class=n>pipeline</span> <span class=o>=</span> <span class=o>...</span> |
| </span></span><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>dataset</span> <span class=o>=</span> <span class=o>..</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>mapped</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl><span class=n>FlatMap</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"FlatMap1"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>dataset</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>String</span> <span class=n>value</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>context</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=n>context</span><span class=o>.</span><span class=na>getCounter</span><span class=o>(</span><span class=s>"my-counter"</span><span class=o>).</span><span class=na>increment</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> <span class=n>context</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>value</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> <span class=o>})</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span></span></span></code></pre></div></div></div><code>MapElements</code> also allows for <code>Context</code> to be accessed by supplying implementations of <code>UnaryFunctionEnv</code> (add second context argument) instead of <code>UnaryFunctor</code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>Pipeline</span> <span class=n>pipeline</span> <span class=o>=</span> <span class=o>...</span> |
| </span></span><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>dataset</span> <span class=o>=</span> <span class=o>...</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>mapped</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>MapElements</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"MapThem"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>dataset</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>input</span><span class=o>,</span> <span class=n>context</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=c1>// use simple counter |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=n>context</span><span class=o>.</span><span class=na>getCounter</span><span class=o>(</span><span class=s>"my-counter"</span><span class=o>).</span><span class=na>increment</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> <span class=k>return</span> <span class=n>input</span><span class=o>.</span><span class=na>toLowerCase</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> <span class=o>})</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span></span></span></code></pre></div></div></div>Accumulators are translated into Beam Metrics in background so they can be viewed the same way. Namespace of translated metrics is set to operator’s name.</p><h3 id=windowing>Windowing</h3><p>Euphoria follows the same <a href=/documentation/programming-guide/#windowing>windowing principles</a> as Beam Java SDK. Every shuffle operator (operator which needs to shuffle data over the network) allows you to set it. The same parameters as in Beam are required. <code>WindowFn</code>, <code>Trigger</code>, <code>WindowingStrategy</code> and other. Users are guided to either set all mandatory and several optional parameters or none when building an operator. Windowing is propagated down through the <code>Pipeline</code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>countedElements</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>CountByKey</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>e</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>windowBy</span><span class=o>(</span><span class=n>FixedWindows</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardSeconds</span><span class=o>(</span><span class=n>1</span><span class=o>)))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>triggeredBy</span><span class=o>(</span><span class=n>DefaultTrigger</span><span class=o>.</span><span class=na>of</span><span class=o>())</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>discardingFiredPanes</span><span class=o>()</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>withAllowedLateness</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>standardSeconds</span><span class=o>(</span><span class=n>5</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>withOnTimeBehavior</span><span class=o>(</span><span class=n>OnTimeBehavior</span><span class=o>.</span><span class=na>FIRE_IF_NON_EMPTY</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>withTimestampCombiner</span><span class=o>(</span><span class=n>TimestampCombiner</span><span class=o>.</span><span class=na>EARLIEST</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span></span></span></code></pre></div></div></div></p><h2 id=how-to-get-euphoria>How to get Euphoria</h2><p>Euphoria is located in <code>dsl-euphoria</code> branch, <code>beam-sdks-java-extensions-euphoria</code> module of The Apache Beam project. To build <code>euphoria</code> subproject call:</p><pre tabindex=0><code>./gradlew beam-sdks-java-extensions-euphoria:build |
| </code></pre><h2 id=operator-reference>Operator Reference</h2><p>Operators are basically higher level data transformations, which allows you to build business logic of your data processing job in a simple way. All the Euphoria operators are documented in this section including examples. There are no examples with <a href=#windowing>windowing</a> applied for the sake of simplicity. Refer to the <a href=#windowing>windowing section</a> for more details.</p><h3 id=countbykey><code>CountByKey</code></h3><p>Counting elements with the same key. Requires input dataset to be mapped by given key extractor (<code>UnaryFunction</code>) to keys which are then counted. Output is emitted as <code>KV<K, Long></code> (<code>K</code> is key type) where each <code>KV</code> contains key and number of element in input dataset for the key.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose input: [1, 2, 4, 1, 1, 3] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>output</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>CountByKey</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>e</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// Output will contain: [KV(1, 3), KV(2, 1), KV(3, 1), (4, 1)] |
| </span></span></span></code></pre></div></div></div></p><h3 id=distinct><code>Distinct</code></h3><p>Outputting distinct (based on equals method) elements. It takes optional <code>UnaryFunction</code> mapper parameter which maps elements to output type.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose input: [1, 2, 3, 3, 2, 1] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>Distinct</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"unique-integers-only"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// Output will contain: 1, 2, 3 |
| </span></span></span><span class=line><span class=cl><span class=c1></span> </span></span></code></pre></div></div></div><code>Distinct</code> with mapper.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose keyValueInput: [KV(1, 100L), KV(3, 100_000L), KV(42, 10L), KV(1, 0L), KV(3, 0L)] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>Distinct</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"unique-keys-only"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>keyValueInput</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>projected</span><span class=o>(</span><span class=n>KV</span><span class=o>::</span><span class=n>getKey</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// Output will contain kvs with keys: 1, 3, 42 with some arbitrary values associated with given keys |
| </span></span></span></code></pre></div></div></div></p><h3 id=join><code>Join</code></h3><p>Represents inner join of two (left and right) datasets on given key producing a new dataset. Key is extracted from both datasets by separate extractors so elements in left and right can have different types denoted as <code>LeftT</code> and <code>RightT</code>. The join itself is performed by user-supplied <code>BinaryFunctor</code> which consumes elements from both dataset sharing the same key. And outputs result of the join (<code>OutputT</code>). The operator emits output dataset of <code>KV<K, OutputT></code> type.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| </span></span></span><span class=line><span class=cl><span class=c1>// suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>String</span><span class=o>>></span> <span class=n>joined</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>Join</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"join-length-to-words"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>left</span><span class=o>,</span> <span class=n>right</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>by</span><span class=o>(</span><span class=n>le</span> <span class=o>-></span> <span class=n>le</span><span class=o>,</span> <span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// key extractors |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>using</span><span class=o>((</span><span class=n>Integer</span> <span class=n>l</span><span class=o>,</span> <span class=n>String</span> <span class=n>r</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>c</span><span class=o>)</span> <span class=o>-></span> <span class=n>c</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>l</span> <span class=o>+</span> <span class=s>"+"</span> <span class=o>+</span> <span class=n>r</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// joined will contain: [ KV(1, "1+X"), KV(3, "3+cat"), KV(3, "3+rat"), KV(4, "4+duck"), |
| </span></span></span><span class=line><span class=cl><span class=c1>// KV(3, "3+cat"), KV(3, "3+rat"), KV(1, "1+X")] |
| </span></span></span></code></pre></div></div></div></p><h3 id=leftjoin><code>LeftJoin</code></h3><p>Represents left join of two (left and right) datasets on given key producing single new dataset. Key is extracted from both datasets by separate extractors so elements in left and right can have different types denoted as <code>LeftT</code> and <code>RightT</code>. The join itself is performed by user-supplied <code>BinaryFunctor</code> which consumes one element from both dataset, where right is present optionally, sharing the same key. And outputs result of the join (<code>OutputT</code>). The operator emits output dataset of <code>KV<K, OutputT></code> type.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| </span></span></span><span class=line><span class=cl><span class=c1>// suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>String</span><span class=o>>></span> <span class=n>joined</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>LeftJoin</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"left-join-length-to-words"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>left</span><span class=o>,</span> <span class=n>right</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>by</span><span class=o>(</span><span class=n>le</span> <span class=o>-></span> <span class=n>le</span><span class=o>,</span> <span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// key extractors |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>Integer</span> <span class=n>l</span><span class=o>,</span> <span class=n>Optional</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>r</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>c</span><span class=o>)</span> <span class=o>-></span> |
| </span></span><span class=line><span class=cl> <span class=n>c</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>l</span> <span class=o>+</span> <span class=s>"+"</span> <span class=o>+</span> <span class=n>r</span><span class=o>.</span><span class=na>orElse</span><span class=o>(</span><span class=kc>null</span><span class=o>)))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// joined will contain: [KV(1, "1+X"), KV(2, "2+null"), KV(3, "3+cat"), |
| </span></span></span><span class=line><span class=cl><span class=c1>// KV(3, "3+rat"), KV(0, "0+null"), KV(4, "4+duck"), KV(3, "3+cat"), |
| </span></span></span><span class=line><span class=cl><span class=c1>// KV(3, "3+rat"), KV(1, "1+X")] |
| </span></span></span></code></pre></div></div></div>Euphoria support performance optimization called ‘BroadcastHashJoin’ for the <code>LeftJoin</code>. Broadcast join can be very efficient when joining two datasets where one fits in memory (in <code>LeftJoin</code> right dataset has to fit in memory). How to use ‘Broadcast Hash Join’ is described in <a href=#translation>Translation</a> section.</p><h3 id=rightjoin><code>RightJoin</code></h3><p>Represents right join of two (left and right) datasets on given key producing single new dataset. Key is extracted from both datasets by separate extractors so elements in left and right can have different types denoted as <code>LeftT</code> and <code>RightT</code>. The join itself is performed by user-supplied <code>BinaryFunctor</code> which consumes one element from both dataset, where left is present optionally, sharing the same key. And outputs result of the join (<code>OutputT</code>). The operator emits output dataset of <code>KV<K, OutputT></code> type.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| </span></span></span><span class=line><span class=cl><span class=c1>// suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>String</span><span class=o>>></span> <span class=n>joined</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>RightJoin</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"right-join-length-to-words"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>left</span><span class=o>,</span> <span class=n>right</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>by</span><span class=o>(</span><span class=n>le</span> <span class=o>-></span> <span class=n>le</span><span class=o>,</span> <span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// key extractors |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>Optional</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>l</span><span class=o>,</span> <span class=n>String</span> <span class=n>r</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>c</span><span class=o>)</span> <span class=o>-></span> |
| </span></span><span class=line><span class=cl> <span class=n>c</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>l</span><span class=o>.</span><span class=na>orElse</span><span class=o>(</span><span class=kc>null</span><span class=o>)</span> <span class=o>+</span> <span class=s>"+"</span> <span class=o>+</span> <span class=n>r</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> <span class=c1>// joined will contain: [ KV(1, "1+X"), KV(3, "3+cat"), KV(3, "3+rat"), |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=c1>// KV(4, "4+duck"), KV(3, "3+cat"), KV(3, "3+rat"), KV(1, "1+X"), |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=c1>// KV(8, "null+elephant"), KV(5, "null+mouse")] |
| </span></span></span></code></pre></div></div></div>Euphoria support performance optimization called ‘BroadcastHashJoin’ for the <code>RightJoin</code>. Broadcast join can be very efficient when joining two datasets where one fits in memory (in <code>RightJoin</code> left dataset has to fit in memory). How to use ‘Broadcast Hash Join’ is described in <a href=#translation>Translation</a> section.</p><h3 id=fulljoin><code>FullJoin</code></h3><p>Represents full outer join of two (left and right) datasets on given key producing single new dataset. Key is extracted from both datasets by separate extractors so elements in left and right can have different types denoted as <code>LeftT</code> and <code>RightT</code>. The join itself is performed by user-supplied <code>BinaryFunctor</code> which consumes one element from both dataset, where both are present only optionally, sharing the same key. And outputs result of the join (<code>OutputT</code>). The operator emits output dataset of <code>KV<K, OutputT></code> type.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose that left contains: [1, 2, 3, 0, 4, 3, 1] |
| </span></span></span><span class=line><span class=cl><span class=c1>// suppose that right contains: ["mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>String</span><span class=o>>></span> <span class=n>joined</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>FullJoin</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"join-length-to-words"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>left</span><span class=o>,</span> <span class=n>right</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>by</span><span class=o>(</span><span class=n>le</span> <span class=o>-></span> <span class=n>le</span><span class=o>,</span> <span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// key extractors |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>Optional</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>l</span><span class=o>,</span> <span class=n>Optional</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>r</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>c</span><span class=o>)</span> <span class=o>-></span> |
| </span></span><span class=line><span class=cl> <span class=n>c</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>l</span><span class=o>.</span><span class=na>orElse</span><span class=o>(</span><span class=kc>null</span><span class=o>)</span> <span class=o>+</span> <span class=s>"+"</span> <span class=o>+</span> <span class=n>r</span><span class=o>.</span><span class=na>orElse</span><span class=o>(</span><span class=kc>null</span><span class=o>)))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// joined will contain: [ KV(1, "1+X"), KV(2, "2+null"), KV(3, "3+cat"), KV(3, "3+rat"), |
| </span></span></span><span class=line><span class=cl><span class=c1>// KV(0, "0+null"), KV(4, "4+duck"), KV(3, "3+cat"), KV(3, "3+rat"),KV(1, "1+X"), |
| </span></span></span><span class=line><span class=cl><span class=c1>// KV(1, "null+elephant"), KV(5, "null+mouse")] |
| </span></span></span></code></pre></div></div></div></p><h3 id=mapelements><code>MapElements</code></h3><p>Transforms one input element of input type <code>InputT</code> to one output element of another (potentially the same) <code>OutputT</code> type. Transformation is done through user specified <code>UnaryFunction</code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose inputs contains: [ 0, 1, 2, 3, 4, 5] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>strings</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>MapElements</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"int2str"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span><span class=n>i</span> <span class=o>-></span> <span class=s>"#"</span> <span class=o>+</span> <span class=n>i</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// strings will contain: [ "#0", "#1", "#2", "#3", "#4", "#5"] |
| </span></span></span></code></pre></div></div></div></p><h3 id=flatmap><code>FlatMap</code></h3><p>Transforms one input element of input type <code>InputT</code> to zero or more output elements of another (potentially the same) <code>OutputT</code> type. Transformation is done through user specified <code>UnaryFunctor</code>, where <code>Collector<OutputT></code> is utilized to emit output elements. Notice similarity with <code>MapElements</code> which can always emit only one element.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose words contain: ["Brown", "fox", ".", ""] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>letters</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>FlatMap</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"str2char"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>words</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>String</span> <span class=n>s</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>collector</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=k>for</span> <span class=o>(</span><span class=kt>int</span> <span class=n>i</span> <span class=o>=</span> <span class=n>0</span><span class=o>;</span> <span class=n>i</span> <span class=o><</span> <span class=n>s</span><span class=o>.</span><span class=na>length</span><span class=o>();</span> <span class=n>i</span><span class=o>++)</span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=kt>char</span> <span class=n>c</span> <span class=o>=</span> <span class=n>s</span><span class=o>.</span><span class=na>charAt</span><span class=o>(</span><span class=n>i</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> <span class=n>collector</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>String</span><span class=o>.</span><span class=na>valueOf</span><span class=o>(</span><span class=n>c</span><span class=o>));</span> |
| </span></span><span class=line><span class=cl> <span class=o>}</span> |
| </span></span><span class=line><span class=cl> <span class=o>})</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// characters will contain: ["B", "r", "o", "w", "n", "f", "o", "x", "."] |
| </span></span></span></code></pre></div></div></div><code>FlatMap</code> may be used to determine time-stamp of elements. It is done by supplying implementation of <code>ExtractEventTime</code> time extractor when building it. There is specialized <code>AssignEventTime</code> operator to assign time-stamp to elements. Consider using it, you code may be more readable.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose events contain events of SomeEventObject, its 'getEventTimeInMillis()' methods returns time-stamp |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>SomeEventObject</span><span class=o>></span> <span class=n>timeStampedEvents</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>FlatMap</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"extract-event-time"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>events</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span> <span class=o>(</span><span class=n>SomeEventObject</span> <span class=n>e</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>SomeEventObject</span><span class=o>></span> <span class=n>c</span><span class=o>)</span> <span class=o>-></span> <span class=n>c</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>e</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>eventTimeBy</span><span class=o>(</span><span class=n>SomeEventObject</span><span class=o>::</span><span class=n>getEventTimeInMillis</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>//Euphoria will now know event time for each event |
| </span></span></span></code></pre></div></div></div></p><h3 id=filter><code>Filter</code></h3><p><code>Filter</code> throws away all the elements which do not pass given condition. The condition is supplied by the user as implementation of <code>UnaryPredicate</code>. Input and output elements are of the same type.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose nums contains: [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>divisibleBythree</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>Filter</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"divisibleByThree"</span><span class=o>).</span><span class=na>of</span><span class=o>(</span><span class=n>nums</span><span class=o>).</span><span class=na>by</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>e</span> <span class=o>%</span> <span class=n>3</span> <span class=o>==</span> <span class=n>0</span><span class=o>).</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>//divisibleBythree will contain: [ 0, 3, 6, 9] |
| </span></span></span></code></pre></div></div></div></p><h3 id=reducebykey><code>ReduceByKey</code></h3><p>Performs aggregation of <code>InputT</code> type elements with the same key through user-supplied reduce function. Key is extracted from each element through <code>UnaryFunction</code> which takes input element and outputs its key of type <code>K</code>. Elements can optionally be mapped to value of type <code>V</code>, it happens before elements shuffle, so it can have positive performance influence.</p><p>Finally, elements with the same key are aggregated by user-defined <code>ReduceFunctor</code>, <code>ReduceFunction</code> or <code>CombinableReduceFunction</code>. They differs in number of arguments they take and in way output is interpreted. <code>ReduceFunction</code> is basically a function which takes <code>Stream</code> of elements as input and outputs one aggregation result. <code>ReduceFunctor</code> takes second <code>Collector</code> which allows for access to <code>Context</code>. When <code>CombinableReduceFunction</code> is provided, partial reduction is performed before shuffle so less data have to be transported through network.</p><p>Following example shows basic usage of <code>ReduceByKey</code> operator including value extraction.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>countOfAnimalNamesByLength</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>ReduceByKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"to-letters-counts"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>animals</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// length of animal name will be used as grouping key |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=c1>// we need to count each animal name once, so why not to optimize each string to 1 |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>valueBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>1</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>reduceBy</span><span class=o>(</span><span class=n>Stream</span><span class=o>::</span><span class=n>count</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// countOfAnimalNamesByLength wil contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L) ] |
| </span></span></span></code></pre></div></div></div></p><p>Now suppose that we want to track our <code>ReduceByKey</code> internals using counter.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>countOfAnimalNamesByLength</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>ReduceByKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"to-letters-couts"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>animals</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// length of animal name will be used as grouping key |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=c1>// we need to count each animal name once, so why not to optimize each string to 1 |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>valueBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>1</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>reduceBy</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>Stream</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>s</span><span class=o>,</span> <span class=n>Collector</span><span class=o><</span><span class=n>Long</span><span class=o>></span> <span class=n>collector</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=n>collector</span><span class=o>.</span><span class=na>collect</span><span class=o>(</span><span class=n>s</span><span class=o>.</span><span class=na>count</span><span class=o>());</span> |
| </span></span><span class=line><span class=cl> <span class=n>collector</span><span class=o>.</span><span class=na>asContext</span><span class=o>().</span><span class=na>getCounter</span><span class=o>(</span><span class=s>"num-of-keys"</span><span class=o>).</span><span class=na>increment</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> <span class=o>})</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// countOfAnimalNamesByLength wil contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L) ] |
| </span></span></span></code></pre></div></div></div></p><p>Again the same example with optimized combinable output.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>countOfAnimalNamesByLength</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>ReduceByKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"to-letters-couts"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>animals</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// length of animal name will e used as grouping key |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=c1>// we need to count each animal name once, so why not to optimize each string to 1 |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>valueBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>1L</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>combineBy</span><span class=o>(</span><span class=n>s</span> <span class=o>-></span> <span class=n>s</span><span class=o>.</span><span class=na>mapToLong</span><span class=o>(</span><span class=n>l</span> <span class=o>-></span> <span class=n>l</span><span class=o>).</span><span class=na>sum</span><span class=o>())</span> <span class=c1>//Stream::count will not be enough |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// countOfAnimalNamesByLength wil contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L) ] |
| </span></span></span></code></pre></div></div></div>Note that the provided <code>CombinableReduceFunction</code> has to be associative and commutative to be truly combinable. So it can be used to compute partial results before shuffle. And then merge partial result to one. That is why simple <code>Stream::count</code> will not work in this example unlike in the previous one.</p><p>Euphoria aims to make code easy to write and read. Therefore some support to write combinable reduce functions in form of <code>Fold</code> or folding function is already there. It allows user to supply only the reduction logic (<code>BinaryFunction</code>) and creates <code>CombinableReduceFunction</code> out of it. Supplied <code>BinaryFunction</code> still have to be associative.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose animals contains : [ "mouse", "rat", "elephant", "cat", "X", "duck"] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>countOfAnimalNamesByLenght</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>ReduceByKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"to-letters-couts"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>animals</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// length of animal name will be used as grouping key |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=c1>// we need to count each animal name once, so why not to optimize each string to 1 |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>valueBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>1L</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>combineBy</span><span class=o>(</span><span class=n>Fold</span><span class=o>.</span><span class=na>of</span><span class=o>((</span><span class=n>l1</span><span class=o>,</span> <span class=n>l2</span><span class=o>)</span> <span class=o>-></span> <span class=n>l1</span> <span class=o>+</span> <span class=n>l2</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// countOfAnimalNamesByLength will contain [ KV.of(1, 1L), KV.of(3, 2L), KV.of(4, 1L), KV.of(5, 1L), KV.of(8, 1L) ] |
| </span></span></span></code></pre></div></div></div></p><h3 id=reducewindow><code>ReduceWindow</code></h3><p>Reduces all elements in a <a href=#windowing>window</a>. The operator corresponds to <code>ReduceByKey</code> with the same key for all elements, so the actual key is defined only by window.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose input contains [ 1, 2, 3, 4, 5, 6, 7, 8 ] |
| </span></span></span><span class=line><span class=cl><span class=c1>//lets assign time-stamp to each input element |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>withEventTime</span> <span class=o>=</span> <span class=n>AssignEventTime</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>).</span><span class=na>using</span><span class=o>(</span><span class=n>i</span> <span class=o>-></span> <span class=n>1000L</span> <span class=o>*</span> <span class=n>i</span><span class=o>).</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl> |
| </span></span><span class=line><span class=cl><span class=n>PCollection</span><span class=o><</span><span class=n>Integer</span><span class=o>></span> <span class=n>output</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>ReduceWindow</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>withEventTime</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>combineBy</span><span class=o>(</span><span class=n>Fold</span><span class=o>.</span><span class=na>of</span><span class=o>((</span><span class=n>i1</span><span class=o>,</span> <span class=n>i2</span><span class=o>)</span> <span class=o>-></span> <span class=n>i1</span> <span class=o>+</span> <span class=n>i2</span><span class=o>))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>windowBy</span><span class=o>(</span><span class=n>FixedWindows</span><span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>Duration</span><span class=o>.</span><span class=na>millis</span><span class=o>(</span><span class=n>5000</span><span class=o>)))</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>triggeredBy</span><span class=o>(</span><span class=n>DefaultTrigger</span><span class=o>.</span><span class=na>of</span><span class=o>())</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>discardingFiredPanes</span><span class=o>()</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>//output will contain: [ 10, 26 ] |
| </span></span></span></code></pre></div></div></div></p><h3 id=sumbykey><code>SumByKey</code></h3><p>Summing elements with same key. Requires input dataset to be mapped by given key extractor (<code>UnaryFunction</code>) to keys. By value extractor, also <code>UnaryFunction</code> which outputs to <code>Long</code>, to values. Those values are then grouped by key and summed. Output is emitted as <code>KV<K, Long></code> (<code>K</code> is key type) where each <code>KV</code> contains key and number of element in input dataset for the key.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose input contains: [ 1, 2, 3, 4, 5, 6, 7, 8, 9 ] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>KV</span><span class=o><</span><span class=n>Integer</span><span class=o>,</span> <span class=n>Long</span><span class=o>>></span> <span class=n>output</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>SumByKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"sum-odd-and-even"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>input</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=n>e</span> <span class=o>%</span> <span class=n>2</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>valueBy</span><span class=o>(</span><span class=n>e</span> <span class=o>-></span> <span class=o>(</span><span class=kt>long</span><span class=o>)</span> <span class=n>e</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// output will contain: [ KV.of(0, 20L), KV.of(1, 25L)] |
| </span></span></span></code></pre></div></div></div></p><h3 id=union><code>Union</code></h3><p>Merge of at least two datasets of the same type without any guarantee about elements ordering.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>//suppose cats contains: [ "cheetah", "cat", "lynx", "jaguar" ] |
| </span></span></span><span class=line><span class=cl><span class=c1>//suppose rodents contains: [ "squirrel", "mouse", "rat", "lemming", "beaver" ] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>String</span><span class=o>></span> <span class=n>animals</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>Union</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"to-animals"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>cats</span><span class=o>,</span> <span class=n>rodents</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>// animal will contain: "cheetah", "cat", "lynx", "jaguar", "squirrel", "mouse", "rat", "lemming", "beaver" |
| </span></span></span></code></pre></div></div></div></p><h3 id=topperkey><code>TopPerKey</code></h3><p>Emits one top-rated element per key. Key of type <code>K</code> is extracted by given <code>UnaryFunction</code>. Another <code>UnaryFunction</code> extractor allows for conversion input elements to values of type <code>V</code>. Selection of top element is based on <em>score</em>, which is obtained from each element by user supplied <code>UnaryFunction</code> called score calculator. Score type is denoted as <code>ScoreT</code> and it is required to extend <code>Comparable<ScoreT></code> so scores of two elements can be compared directly. Output dataset elements are of type <code>Triple<K, V, ScoreT></code>.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose 'animals contain: [ "mouse", "elk", "rat", "mule", "elephant", "dinosaur", "cat", "duck", "caterpillar" ] |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>Triple</span><span class=o><</span><span class=n>Character</span><span class=o>,</span> <span class=n>String</span><span class=o>,</span> <span class=n>Integer</span><span class=o>>></span> <span class=n>longestNamesByLetter</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>TopPerKey</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"longest-animal-names"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>animals</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>keyBy</span><span class=o>(</span><span class=n>name</span> <span class=o>-></span> <span class=n>name</span><span class=o>.</span><span class=na>charAt</span><span class=o>(</span><span class=n>0</span><span class=o>))</span> <span class=c1>// first character is the key |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>valueBy</span><span class=o>(</span><span class=n>UnaryFunction</span><span class=o>.</span><span class=na>identity</span><span class=o>())</span> <span class=c1>// value type is the same as input element type |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>scoreBy</span><span class=o>(</span><span class=n>String</span><span class=o>::</span><span class=n>length</span><span class=o>)</span> <span class=c1>// length defines score, note that Integer implements Comparable<Integer> |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>//longestNamesByLetter wil contain: [ ('m', "mouse", 5), ('r', "rat", 3), ('e', "elephant", 8), ('d', "dinosaur", 8), ('c', "caterpillar", 11) ] |
| </span></span></span></code></pre></div></div></div><code>TopPerKey</code> is a shuffle operator so it allows for widowing to be defined.</p><h3 id=assigneventtime><code>AssignEventTime</code></h3><p>Euphoria needs to know how to extract time-stamp from elements when <a href=#windowing>windowing</a> is applied. <code>AssignEventTime</code> tells Euphoria how to do that through given implementation of <code>ExtractEventTime</code> function.<div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=c1>// suppose events contain events of SomeEventObject, its 'getEventTimeInMillis()' methods returns time-stamp |
| </span></span></span><span class=line><span class=cl><span class=c1></span><span class=n>PCollection</span><span class=o><</span><span class=n>SomeEventObject</span><span class=o>></span> <span class=n>timeStampedEvents</span> <span class=o>=</span> |
| </span></span><span class=line><span class=cl> <span class=n>AssignEventTime</span><span class=o>.</span><span class=na>named</span><span class=o>(</span><span class=s>"extract-event-time"</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>of</span><span class=o>(</span><span class=n>events</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>using</span><span class=o>(</span><span class=n>SomeEventObject</span><span class=o>::</span><span class=n>getEventTimeInMillis</span><span class=o>)</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>output</span><span class=o>();</span> |
| </span></span><span class=line><span class=cl><span class=c1>//Euphoria will now know event time for each event |
| </span></span></span></code></pre></div></div></div></p><h2 id=translation>Translation</h2><p>Euphoria API is built on top of Beam Java SDK. The API is transparently translated into Beam’s <code>PTransforms</code> in background.</p><p>The fact that Euphoria API is translated to Beam Java SDK give us option to fine tune the translation itself. Translation of an <code>Operator</code> is realized through implementations of <code>OperatorTranslator</code>. |
| Euphoria uses <code>TranslationProvider</code> to decide which translator should be used. User of Euphoria API can supply its own <code>OperatorTranslator</code> through <code>TranslationProvider</code> by extending <code>EuphoriaOptions</code>. |
| Euphoria already contains some useful implementations.</p><h3 id=translationproviders>TranslationProviders</h3><h4 id=generictranslatorprovider><code>GenericTranslatorProvider</code></h4><p>General <code>TranslationProvider</code>. Allows for registration of <code>OperatorTranslator</code> three different ways:</p><ul><li>Registration of operator specific translator by operator class.</li><li>Registration operator specific translator by operator class and additional user defined predicate.</li><li>Registration of general (not specific to one operator type) translator with user defined predicate. |
| Order of registration is important since <code>GenericTranslatorProvider</code> returns first suitable translator.</li></ul><div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>GenericTranslatorProvider</span><span class=o>.</span><span class=na>newBuilder</span><span class=o>()</span> |
| </span></span><span class=line><span class=cl> <span class=o>.</span><span class=na>register</span><span class=o>(</span><span class=n>FlatMap</span><span class=o>.</span><span class=na>class</span><span class=o>,</span> <span class=k>new</span> <span class=n>FlatMapTranslator</span><span class=o><>())</span> <span class=c1>// register by operator class |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>register</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=n>Join</span><span class=o>.</span><span class=na>class</span><span class=o>,</span> |
| </span></span><span class=line><span class=cl> <span class=o>(</span><span class=n>Join</span> <span class=n>op</span><span class=o>)</span> <span class=o>-></span> <span class=o>{</span> |
| </span></span><span class=line><span class=cl> <span class=n>String</span> <span class=n>name</span> <span class=o>=</span> <span class=o>((</span><span class=n>Optional</span><span class=o><</span><span class=n>String</span><span class=o>>)</span> <span class=n>op</span><span class=o>.</span><span class=na>getName</span><span class=o>()).</span><span class=na>orElse</span><span class=o>(</span><span class=s>""</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> <span class=k>return</span> <span class=n>name</span><span class=o>.</span><span class=na>toLowerCase</span><span class=o>().</span><span class=na>startsWith</span><span class=o>(</span><span class=s>"broadcast"</span><span class=o>);</span> |
| </span></span><span class=line><span class=cl> <span class=o>},</span> |
| </span></span><span class=line><span class=cl> <span class=k>new</span> <span class=n>BroadcastHashJoinTranslator</span><span class=o><>())</span> <span class=c1>// register by class and predicate |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>register</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=n>op</span> <span class=o>-></span> <span class=n>op</span> <span class=k>instanceof</span> <span class=n>CompositeOperator</span><span class=o>,</span> |
| </span></span><span class=line><span class=cl> <span class=k>new</span> <span class=n>CompositeOperatorTranslator</span><span class=o><>())</span> <span class=c1>// register by predicate only |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=o>.</span><span class=na>build</span><span class=o>();</span></span></span></code></pre></div></div></div><p><code>GenericTranslatorProvider</code> is default provider, see <code>GenericTranslatorProvider.createWithDefaultTranslators()</code>.</p><h4 id=compositeprovider><code>CompositeProvider</code></h4><p>Implements chaining of <code>TranslationProvider</code>s in given order. That in turn allows for composing user defined <code>TranslationProvider</code> with already supplied by Euphoria API.</p><div class='language-java snippet'><div class="notebook-skip code-snippet"><a class=copy type=button data-bs-toggle=tooltip data-bs-placement=bottom title="Copy to clipboard"><img src=/images/copy-icon.svg></a><div class=highlight><pre tabindex=0 class=chroma><code class=language-java data-lang=java><span class=line><span class=cl><span class=n>CompositeProvider</span><span class=o>.</span><span class=na>of</span><span class=o>(</span> |
| </span></span><span class=line><span class=cl> <span class=n>CustomTranslatorProvider</span><span class=o>.</span><span class=na>of</span><span class=o>(),</span> <span class=c1>// first ask CustomTranslatorProvider for translator |
| </span></span></span><span class=line><span class=cl><span class=c1></span> <span class=n>GenericTranslatorProvider</span><span class=o>.</span><span class=na>createWithDefaultTranslators</span><span class=o>());</span> <span class=c1>// then ask default provider if needed |
| </span></span></span></code></pre></div></div></div><h3 id=operator-translators>Operator Translators</h3><p>Each <code>Operator</code> needs to be translated to Java Beam SDK. That is done by implementations of <code>OperatorTranslator</code>. Euphoria API contains translator for every <code>Operator</code> implementation supplied with it. |
| Some operators may have an alternative translations suitable in some cases. <code>Join</code> typically may have many implementations. We are describing only the most interesting here.</p><h4 id=broadcasthashjointranslator><code>BroadcastHashJoinTranslator</code></h4><p>Is able to translate <code>LeftJoin</code> and <code>RightJoin</code> when whole dataset of one side fits in memory of target executor. So it can be distributed using Beam’s side inputs. Resulting in better performance.</p><h4 id=compositeoperatortranslator><code>CompositeOperatorTranslator</code></h4><p>Some operators are composite. Meaning that they are in fact wrapped chain of other operators. <code>CompositeOperatorTranslator</code> ensures that they are decomposed to elemental operators during translation process.</p><h3 id=details>Details</h3><p>Most of the translation happens in <code>org.apache.beam.sdk.extensions.euphoria.core.translate</code> package. Where the most interesting classes are:</p><ul><li><code>OperatorTranslator</code> - Interface which defining inner API of Euphoria to Beam translation.</li><li><code>TranslatorProvider</code> - Way of supplying custom translators.</li><li><code>OperatorTransform</code> - Is governing actual translation and/or expansion Euphoria’s operators to Beam’s <code>PTransform</code></li><li><code>EuphoriaOptions</code> - A <code>PipelineOptions</code>, allows for setting custom <code>TranslatorProvider</code>.</li></ul><p>The package also contains implementation of <code>OperatorTranslator</code> for each supported operator type (<code>JoinTranslator</code>, <code>FlatMapTranslator</code>, <code>ReduceByKeyTranslator</code>). Not every operator needs to have translator of its own. Some of them can be composed from other operators. That is why operators may implement <code>CompositeOperator</code> which give them option to be expanded to set of other Euphoria operators.</p><p>The translation process was designed with flexibility in mind. We wanted to allow different ways of translating higher-level Euphoria operators to Beam’s SDK’s primitives. It allows for further performance optimizations based on user choices or some knowledge about data obtained automatically.</p><h2 id=unsupported-features>Unsupported Features</h2><p><a href=https://github.com/seznam/euphoria>Original Euphoria</a> contained some features and operators not jet supported in Beam port. List of not yet supported features follows:</p><ul><li><code>ReduceByKey</code> in original Euphoria was allowed to sort output values (per key). This is also not yet translatable into Beam, therefore not supported.</li></ul></div></div><footer class=footer><div class=footer__contained><div class=footer__cols><div class="footer__cols__col footer__cols__col__logos"><div class=footer__cols__col__logo><img src=/images/beam_logo_circle.svg class=footer__logo alt="Beam logo"></div><div class=footer__cols__col__logo><img src=/images/apache_logo_circle.svg class=footer__logo alt="Apache logo"></div></div><div class=footer-wrapper><div class=wrapper-grid><div class=footer__cols__col><div class=footer__cols__col__title>Start</div><div class=footer__cols__col__link><a href=/get-started/beam-overview/>Overview</a></div><div class=footer__cols__col__link><a href=/get-started/quickstart-java/>Quickstart (Java)</a></div><div class=footer__cols__col__link><a href=/get-started/quickstart-py/>Quickstart (Python)</a></div><div class=footer__cols__col__link><a href=/get-started/quickstart-go/>Quickstart (Go)</a></div><div class=footer__cols__col__link><a href=/get-started/downloads/>Downloads</a></div></div><div class=footer__cols__col><div class=footer__cols__col__title>Docs</div><div class=footer__cols__col__link><a href=/documentation/programming-guide/>Concepts</a></div><div class=footer__cols__col__link><a href=/documentation/pipelines/design-your-pipeline/>Pipelines</a></div><div class=footer__cols__col__link><a href=/documentation/runners/capability-matrix/>Runners</a></div></div><div class=footer__cols__col><div class=footer__cols__col__title>Community</div><div class=footer__cols__col__link><a href=/contribute/>Contribute</a></div><div class=footer__cols__col__link><a href=https://projects.apache.org/committee.html?beam target=_blank>Team<img src=/images/external-link-icon.png width=14 height=14 alt="External link."></a></div><div class=footer__cols__col__link><a href=/community/presentation-materials/>Media</a></div><div class=footer__cols__col__link><a href=/community/in-person/>Events/Meetups</a></div><div class=footer__cols__col__link><a href=/community/contact-us/>Contact Us</a></div></div><div class=footer__cols__col><div class=footer__cols__col__title>Resources</div><div class=footer__cols__col__link><a href=/blog/>Blog</a></div><div class=footer__cols__col__link><a href=https://github.com/apache/beam>GitHub</a></div></div></div><div class=footer__bottom>© |
| <a href=https://www.apache.org>The Apache Software Foundation</a> |
| | <a href=/privacy_policy>Privacy Policy</a> |
| | <a href=/feed.xml>RSS Feed</a><br><br>Apache Beam, Apache, Beam, the Beam logo, and the Apache feather logo are either registered trademarks or trademarks of The Apache Software Foundation. All other products or name brands are trademarks of their respective holders, including The Apache Software Foundation.</div></div><div class="footer__cols__col footer__cols__col__logos"><div class=footer__cols__col--group><div class=footer__cols__col__logo><a href=https://github.com/apache/beam><img src=/images/logos/social-icons/github-logo-150.png class=footer__logo alt="Github logo"></a></div><div class=footer__cols__col__logo><a href=https://www.linkedin.com/company/apache-beam/><img src=/images/logos/social-icons/linkedin-logo-150.png class=footer__logo alt="Linkedin logo"></a></div></div><div class=footer__cols__col--group><div class=footer__cols__col__logo><a href=https://twitter.com/apachebeam><img src=/images/logos/social-icons/twitter-logo-150.png class=footer__logo alt="Twitter logo"></a></div><div class=footer__cols__col__logo><a href=https://www.youtube.com/channel/UChNnb_YO_7B0HlW6FhAXZZQ><img src=/images/logos/social-icons/youtube-logo-150.png class=footer__logo alt="Youtube logo"></a></div></div></div></div></div></footer></body></html> |
