An exclusive Inside Climate News analysis found companies have spilled nearly 150 million gallons of toxic, highly saline wastewater in Texas over the last decade.

Oil and gas companies spill millions of gallons of wastewater in Texas

Milestone Environmental Services said it planned to lease about 22,000 acres in Midland and Loving counties in Texas.

Permian Basin energy companies focusing on carbon capture during region's growth

An exclusive Inside Climate News analysis found that companies have spilled nearly 150 million gallons of toxic, highly saline wastewater in Texas over the last decade.

WASHINGTON (AP) — The World Bank reported on Monday that oil prices could be pushed into "uncharted waters" if the violence between Israel and Hamas widens, which could result in increased food prices worldwide. The World Bank's Commodity Markets Outlook ...

Oil prices could reach 'uncharted waters' if the Israel-Hamas war escalates, the World Bank says

The U.S. Energy secretary on Monday announced Arizona is set to receive part of a $1.3 billion federal grant to boost the state's power systems in rural areas.             Speaking at the Apache Generating Station in Cochise County,

Arizona will get some of a billion dollar federal energy grant. Here's why

WASHINGTON (AP) — The World Bank reported Monday that oil prices could be pushed into “uncharted waters” if the violence between Israel and Hamas intensifies, which could result in increased food prices worldwide.

Permian Resources ( NYSE: PR ), Chord Energy ( NASDAQ: CHRD ), and Magnolia Oil & Gas ( NYSE: MGY) could be the next acquisition targets in the oil industry as consolidation continues with Exxon Mobil's ( XOM) and Chevron's ( CVX) recent mega transactions.

Permian Resources, Chord Energy, Magnolia Oil may be next M&A targets - analyst

Texas and the Permian Basin set new records for oil and natural gas production in September and the Permian Basin Petroleum Association and Texas Oil & Gas Association say that and the record in-state refineries set in July were accomplished in concert with greater environmental protections.

State pumps record oil, gas in September

A contract worker has died in a water storage pit at a Birchcliff Energy facility in northern Alberta, the company said Wednesday. The Calgary-based oil and gas company says the incident occurred ...

Worker dead in apparent drowning at northern Alberta energy facility

The disconnect between what oil companies and many energy experts think will happen in the coming years has never been quite this stark. Big oil companies are doubling down on drilling for oil and gas and processing it into fuels for use in engines ...

Chasing Big Mergers, Oil Executives Dismiss Peak Oil Concerns

Apache

apache

Handling JSON, Schema Issues, and Doris Quirks in an Apache SeaTunnel Pipeline

Dear Apache SeaTunnel Community, We are excited to announce that Apache SeaTunnel 2.3.10 has been officially released, with @zhangshenghang serving as the release manager! Apache SeaTunnel 2.3.10 @zhangshenghang This update significantly improves connector functionality, enhances configuration options, and fixes various bug fixes, further boosting the stability and performance of data synchronization and processing. These enhancements empower enterprises to build more efficient and flexible data integration platforms. significantly improves connector functionality enhances configuration options various bug fixes efficient and flexible 📥 Download SeaTunnel 2.3.10 Download Link
Release Notes Download Link Download Link Download Link Download Link Release Notes Release Notes Release Notes Release Notes 🔥 New Features and Improvements Key Enhancements Connector-V2 Added predicate pushdown support for Between in the Paimon Connector.
Introduced message tag configuration for RocketMQ Source.
Enabled single-file mode support for File Sink.
Supported multiple StarRocks Sources.
Added an optional flag to skip parsing errors in the RocketMQ Connector. Added predicate pushdown support for Between in the Paimon Connector. predicate pushdown support Between Paimon Connector Introduced message tag configuration for RocketMQ Source. message tag configuration RocketMQ Source Enabled single-file mode support for File Sink. single-file mode File Sink Supported multiple StarRocks Sources. multiple StarRocks Sources Added an optional flag to skip parsing errors in the RocketMQ Connector. optional flag skip parsing errors Transforms-V2 Introduced support for non-standard LLM responses.
Added Zhipu AI support in Embedding and LLM modules.
Improved JSONPath handling for Map Array types. Introduced support for non-standard LLM responses. non-standard LLM responses Added Zhipu AI support in Embedding and LLM modules. Zhipu AI support Embedding LLM modules Improved JSONPath handling for Map Array types. JSONPath handling Map Array types Configuration Enhancements Introduced custom encryption/decryption keys for configurations, offering greater flexibility. Introduced custom encryption/decryption keys for configurations, offering greater flexibility. custom encryption/decryption keys greater flexibility Other Enhancements Optimized connectors, including HBase, Oracle-CDC, Google Sheets, DingTalk, Slack, and more. Optimized connectors, including HBase, Oracle-CDC, Google Sheets, DingTalk, Slack, and more. HBase, Oracle-CDC, Google Sheets, DingTalk, Slack 🛠️ Connector Configuration Optimization Improved configuration options for LocalFile, Web3j, InfluxDB, HTTP, IoTDB, Email, and more to enhance user experience and flexibility.
Strengthened and optimized parameters for StarRocks, JDBC (SQLServer/Dameng), Iceberg, Redis, ClickHouse, Kafka, and others. Improved configuration options for LocalFile, Web3j, InfluxDB, HTTP, IoTDB, Email, and more to enhance user experience and flexibility. LocalFile, Web3j, InfluxDB, HTTP, IoTDB, Email user experience flexibility Strengthened and optimized parameters for StarRocks, JDBC (SQLServer/Dameng), Iceberg, Redis, ClickHouse, Kafka, and others. StarRocks, JDBC (SQLServer/Dameng), Iceberg, Redis, ClickHouse, Kafka Core Module Enhancements Zeta Engine:
Added scheduled log file handle cleanup.
Separated classloader configurations for better modularity.
Deprecated RESTful API v1 for enhanced security and efficiency.
Core Improvements:
Enabled parsing of quotes as keys.
Unified the Engine initialization logic for connectors. Zeta Engine: Zeta Engine: Added scheduled log file handle cleanup. scheduled log file handle cleanup Separated classloader configurations for better modularity. Separated classloader configurations Deprecated RESTful API v1 for enhanced security and efficiency. Deprecated RESTful API v1 Core Improvements: Core Improvements: Enabled parsing of quotes as keys. parsing of quotes as keys Unified the Engine initialization logic for connectors. Engine initialization logic for connectors 🐞 Bug Fixes & Reliability Improvements This release fixes numerous issues, ensuring greater stability in production environments. fixes numerous issues greater stability Key Fixes: Resolved Milvus SourceReader state check errors.
Fixed Kafka source duplicate reading issues.
Addressed CSV file read/write inconsistencies.
Fixed various file handling and network request errors in multiple connectors.
Resolved issues related to Doris, Mongo-CDC, StarRocks, Hive, SQL Server, Oracle, and more to ensure seamless data synchronization.
Optimized logging, plugin discovery mechanisms, and CI workflows for enhanced system robustness. Resolved Milvus SourceReader state check errors. Milvus SourceReader state check errors Fixed Kafka source duplicate reading issues. Kafka source duplicate reading issues Addressed CSV file read/write inconsistencies. CSV file read/write inconsistencies Fixed various file handling and network request errors in multiple connectors. various file handling and network request errors Resolved issues related to Doris, Mongo-CDC, StarRocks, Hive, SQL Server, Oracle, and more to ensure seamless data synchronization. Doris, Mongo-CDC, StarRocks, Hive, SQL Server, Oracle ensure seamless data synchronization Optimized logging, plugin discovery mechanisms, and CI workflows for enhanced system robustness. logging, plugin discovery mechanisms, and CI workflows 📖 Documentation Enhancements Updated documentation for FTP, JDBC, S3File, Oracle, ClickHouse, and more.
Fixed broken links, incorrect parameters, and expanded Chinese translations for better accessibility. Updated documentation for FTP, JDBC, S3File, Oracle, ClickHouse, and more. FTP, JDBC, S3File, Oracle, ClickHouse Fixed broken links, incorrect parameters, and expanded Chinese translations for better accessibility. broken links, incorrect parameters expanded Chinese translations 🎉 Acknowledging Our Contributors Special thanks to @zhangshenhang for leading this release! We also extend our gratitude to the following community contributors for their incredible efforts: @zhangshenhang following community contributors 📢 Summary The Apache SeaTunnel 2.3.10 release brings major improvements in functionality, performance, and stability. We believe these enhancements will significantly improve your data integration and big data processing workflows. Apache SeaTunnel 2.3.10 major improvements in functionality, performance, and stability improve your data integration and big data processing workflows A huge thank you to our community members and contributors for their hard work and dedication! thank you community members and contributors 👉 Download the latest version today and explore the newly upgraded Apache SeaTunnel! Let’s continue to drive innovation in data integration together! Download the latest version today Apache SeaTunnel

Apache SeaTunnel's Latest Update Brings Improved Connector Functionality

Stop Wasting Time on Broken Data Syncs—Here’s How to Do It Right with Apache SeaTunnel

Is Your Apache Ni-Fi Ready for Production?

Lots of service providers offer a free tier of their service. The idea is to let you kick their service's tires freely. If you need to go above the free tier at any point, you'll likely stay on the service and pay. In this day and age, most services are online and accessible via an API. Today, we will implement a free tier with Apache APISIX. A naive approach I implemented a free tier in my post Evolving your RESTful APIs, a step-by-step approach, albeit in a very naive way. I copy-pasted the limit-count plugin and added my required logic. function _M.access(conf, ctx)
    core.log.info("ver: ", ctx.conf_version)

    -- no limit if the request is authenticated
    local key = core.request.header(ctx, conf.header)                               #1
    if key then
        local consumer_conf = consumer_mod.plugin("key-auth")                       #2
        if consumer_conf then
            local consumers = lrucache("consumers_key", consumer_conf.conf_version, #3
                    create_consume_cache, consumer_conf)
            local consumer = consumers[key]                                         #4
            if consumer then                                                        #5
                return
            end
        end
    end
-- rest of the logic Get the configured request header value
Get the consumer's key-auth configuration
Get consumers
Get the consumer with the passed API key if they exist
If they exist, bypass the rate-limiting logic The downside of this approach is that the code is now my own. It has evolved since I copied it, and I'm stuck with the version I copied. We can do better with the help of the vars parameter on routes. APISIX route matching APISIX delegates its matching rule to a router. Standard matching parameters include: The URI
The HTTP method. By default, all methods match.
The host
The remote address Most users do match on the URI; a small minority use HTTP methods and the host. However, they are not the only matching parameters. Knowing the rest will bring you into the world of advanced users of APISIX. Let's take a simple example: header-based API versioning. You'd need actually to match a specific HTTP request header. I've already described how to do it previously. In essence, vars is an additional matching criterion that allows the evaluation of APISIX and nginx variables. routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]] The above route will match if, and only if, the HTTP Accept header is equal to vnd.ch.frankel.myservice.v1+json. You can find the complete list of supported operators in the lua-resty-expr project. APISIX matches routes in a non-specified order by default. If URIs are disjointed, that's not an issue. routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]] Problems arise when URIs are somehow not disjointed. For example, imagine I want to set a default route for unversioned calls. routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]]
  - uri: /*
    upstream_id: 1 We need the third route to be evaluated last. If it's evaluated first, it will match all requests, regardless of their HTTP headers. APISIX offers the priority parameter to order route evaluation. By default, a route's priority is 0. Let's use priority to implement the versioning correctly: routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
    priority: 10                                              #1
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]]
    priority: 10                                              #1
  - uri: /*
    upstream_id: 1 Evaluated first. The order is not relevant since the URIs are disjointed. Implementing a free tier with matching rules We are now ready to implement our free tier with matching rules. The first route to be evaluated should be the one with authentication and no rate limit: routes:
  - uri: /get
    upstream_id: 1
    vars: [[ "http_apikey", "~~", ".*"]]                      #1
    plugins:
      key-auth: ~                                             #2
    priority: 10                                              #3 Match if the request has an HTTP header named apikey
Authenticate the request
Evaluate first The other route is evaluated afterward. - uri: /get
    upstream_id: 1
    plugins:
      limit-count:                                            #1
        count: 1
        time_window: 60
        rejected_code: 429 Rate limit this route. When you configure APISIX with the above snippets, and it receives a request to /get, it tries to match the first route only if it has an apikey request header: If it has one:

The key-auth plugin kicks in
If it succeeds, APISIX forwards the request to the upstream
If it fails, APISIX returns a 403 HTTP status code


If it has no such request header, it matches the second route with a rate limit. Conclusion A free tier is a must for any API service provider worth its salt. In this post, I've explained how to configure such free tier with Apache APISIX. The complete source code for this post can be found on GitHub: https://github.com/ajavageek/free-tier-apisix?embedable=true To go further: Consumer
limit-count plugin
router-radixtree
lua-resty-expr Lots of service providers offer a free tier of their service. The idea is to let you kick their service's tires freely. If you need to go above the free tier at any point, you'll likely stay on the service and pay. In this day and age, most services are online and accessible via an API. Today, we will implement a free tier with Apache APISIX . Apache APISIX A naive approach I implemented a free tier in my post Evolving your RESTful APIs, a step-by-step approach , albeit in a very naive way. I copy-pasted the limit-count plugin and added my required logic. Evolving your RESTful APIs, a step-by-step approach limit-count function _M.access(conf, ctx)
    core.log.info("ver: ", ctx.conf_version)

    -- no limit if the request is authenticated
    local key = core.request.header(ctx, conf.header)                               #1
    if key then
        local consumer_conf = consumer_mod.plugin("key-auth")                       #2
        if consumer_conf then
            local consumers = lrucache("consumers_key", consumer_conf.conf_version, #3
                    create_consume_cache, consumer_conf)
            local consumer = consumers[key]                                         #4
            if consumer then                                                        #5
                return
            end
        end
    end
-- rest of the logic function _M.access(conf, ctx)
    core.log.info("ver: ", ctx.conf_version)

    -- no limit if the request is authenticated
    local key = core.request.header(ctx, conf.header)                               #1
    if key then
        local consumer_conf = consumer_mod.plugin("key-auth")                       #2
        if consumer_conf then
            local consumers = lrucache("consumers_key", consumer_conf.conf_version, #3
                    create_consume_cache, consumer_conf)
            local consumer = consumers[key]                                         #4
            if consumer then                                                        #5
                return
            end
        end
    end
-- rest of the logic Get the configured request header value Get the consumer's key-auth configuration Get consumers Get the consumer with the passed API key if they exist If they exist, bypass the rate-limiting logic Get the configured request header value Get the consumer's key-auth configuration key-auth Get consumers Get the consumer with the passed API key if they exist If they exist, bypass the rate-limiting logic The downside of this approach is that the code is now my own. It has evolved since I copied it, and I'm stuck with the version I copied. We can do better with the help of the vars parameter on routes. vars APISIX route matching APISIX delegates its matching rule to a router . router Standard matching parameters include: The URI The HTTP method. By default, all methods match. The host The remote address The URI The HTTP method. By default, all methods match. The host The remote address Most users do match on the URI; a small minority use HTTP methods and the host. However, they are not the only matching parameters. Knowing the rest will bring you into the world of advanced users of APISIX. Let's take a simple example: header-based API versioning. You'd need actually to match a specific HTTP request header. I've already described how to do it previously . In essence, vars is an additional matching criterion that allows the evaluation of APISIX and nginx variables. previously vars routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]] routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]] The above route will match if, and only if, the HTTP Accept header is equal to vnd.ch.frankel.myservice.v1+json . You can find the complete list of supported operators in the lua-resty-expr project. Accept vnd.ch.frankel.myservice.v1+json lua-resty-expr APISIX matches routes in a non-specified order by default. If URIs are disjointed , that's not an issue. disjointed routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]] routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]] Problems arise when URIs are somehow not disjointed. For example, imagine I want to set a default route for unversioned calls. routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]]
  - uri: /*
    upstream_id: 1 routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]]
  - uri: /*
    upstream_id: 1 We need the third route to be evaluated last. If it's evaluated first, it will match all requests, regardless of their HTTP headers. APISIX offers the priority parameter to order route evaluation. By default, a route's priority is 0. Let's use priority to implement the versioning correctly: priority priority routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
    priority: 10                                              #1
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]]
    priority: 10                                              #1
  - uri: /*
    upstream_id: 1 routes:
  - uri: /*
    upstream_id: 1
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v1+json" ]]
    priority: 10                                              #1
  - uri: /*
    upstream_id: 2
    vars: [[ "http_accept", "==", "vnd.ch.frankel.myservice.v2+json" ]]
    priority: 10                                              #1
  - uri: /*
    upstream_id: 1 Evaluated first. The order is not relevant since the URIs are disjointed. Evaluated first. The order is not relevant since the URIs are disjointed. Implementing a free tier with matching rules We are now ready to implement our free tier with matching rules. The first route to be evaluated should be the one with authentication and no rate limit: routes:
  - uri: /get
    upstream_id: 1
    vars: [[ "http_apikey", "~~", ".*"]]                      #1
    plugins:
      key-auth: ~                                             #2
    priority: 10                                              #3 routes:
  - uri: /get
    upstream_id: 1
    vars: [[ "http_apikey", "~~", ".*"]]                      #1
    plugins:
      key-auth: ~                                             #2
    priority: 10                                              #3 Match if the request has an HTTP header named apikey Authenticate the request Evaluate first Match if the request has an HTTP header named apikey apikey Authenticate the request Evaluate first The other route is evaluated afterward. - uri: /get
    upstream_id: 1
    plugins:
      limit-count:                                            #1
        count: 1
        time_window: 60
        rejected_code: 429 - uri: /get
    upstream_id: 1
    plugins:
      limit-count:                                            #1
        count: 1
        time_window: 60
        rejected_code: 429 Rate limit this route. Rate limit this route. When you configure APISIX with the above snippets, and it receives a request to /get , it tries to match the first route only if it has an apikey request header: /get only apikey If it has one:

The key-auth plugin kicks in
If it succeeds, APISIX forwards the request to the upstream
If it fails, APISIX returns a 403 HTTP status code If it has no such request header, it matches the second route with a rate limit. If it has one: The key-auth plugin kicks in
If it succeeds, APISIX forwards the request to the upstream
If it fails, APISIX returns a 403 HTTP status code The key-auth plugin kicks in If it succeeds, APISIX forwards the request to the upstream If it fails, APISIX returns a 403 HTTP status code The key-auth plugin kicks in key-auth If it succeeds, APISIX forwards the request to the upstream If it fails, APISIX returns a 403 HTTP status code If it has no such request header, it matches the second route with a rate limit. Conclusion A free tier is a must for any API service provider worth its salt. In this post, I've explained how to configure such free tier with Apache APISIX. The complete source code for this post can be found on GitHub: https://github.com/ajavageek/free-tier-apisix?embedable=true https://github.com/ajavageek/free-tier-apisix?embedable=true To go further: To go further: Consumer limit-count plugin router-radixtree lua-resty-expr Consumer Consumer limit-count plugin limit-count plugin router-radixtree router-radixtree lua-resty-expr lua-resty-expr

Implementing a Free Tier API: Route Matching with Apache APISIX

Apache Iceberg is an innovative data lakehouse table format designed to revolutionize how you manage large-scale data across various storage layers, such as Hadoop and AWS S3. By treating these diverse storage solutions as a cohesive, universal database, Apache Iceberg facilitates seamless integration with numerous tools, platforms, and interfaces, enhancing both flexibility and accessibility. One of the standout features of Apache Iceberg is its support for ACID transactions, which ensures data integrity by enabling atomicity, consistency, isolation, and durability in data processing operations. Additionally, Apache Iceberg's Time Travel capability allows users to query historical versions of data, providing valuable insights into data changes over time. The format also supports dynamic schema evolution and partitioning adjustments without downtime, which simplifies the management of data as it grows and changes. Apache Iceberg also includes a host of other advanced features that enhance data usability and management. These include snapshot isolation for concurrent data access, upserts and deletes within tables, and handling large-scale metadata efficiently. To help you harness the full potential of Apache Iceberg, our latest blog series gathers a collection of straightforward, hands-on tutorials. These guides are designed to provide you with practical experience in working with Apache Iceberg, from setting up your first table to executing complex data operations. Whether you're a data scientist, engineer, or analyst, these tutorials offer a valuable opportunity to enhance your data handling skills and leverage Apache Iceberg’s powerful features in your projects. Self-Contained Exercises These are completely self-contained and can be done from your laptop; all infrastructure is spun up as Docker containers. End-to-End Basic Data Engineering Tutorial (Spark, Dremio, Superset)
From Postgres to Dashboards with Dremio and Apache Iceberg
From SQLServer to Dashboards with Dremio and Apache Iceberg
From MongoDB to Dashboards with Dremio and Apache Iceberg
Intro to Dremio, Nessie, and Apache Iceberg on Your Laptop
Using Flink with Apache Iceberg and Nessie
Getting Started with Project Nessie, Apache Iceberg, and Apache Spark Using Docker Require Online Services These are tutorials that will require signing up for services like Upsolver, Dremio, AWS, and others to complete. How to Convert JSON Files Into an Apache Iceberg Table with Dremio
How to Convert CSV Files into an Apache Iceberg table with Dremio
Run Graph Queries on Apache Iceberg Tables with Dremio & Puppygraph
BI Dashboards with Apache Iceberg Using AWS Glue and Apache Superset
Streaming and Batch Data Lakehouses with Apache Iceberg, Dremio and Upsolver
Git for Data with Dremio’s Lakehouse Catalog: Easily Ensure Data Quality in Your Data Lakehouse
How to Create a Lakehouse with Airbyte, S3, Apache Iceberg, and Dremio Video Playlists: Hands-On With Apache Iceberg


Apache Iceberg Lakehouse Engineering 101
In summary, Apache Iceberg is more than just a data storage format; it is a comprehensive solution that adapts to the complexities of modern data environments, ensuring robust data integrity, flexibility, and scalability. Apache Iceberg is an innovative data lakehouse table format designed to revolutionize how you manage large-scale data across various storage layers, such as Hadoop and AWS S3. By treating these diverse storage solutions as a cohesive, universal database, Apache Iceberg facilitates seamless integration with numerous tools, platforms, and interfaces, enhancing both flexibility and accessibility. One of the standout features of Apache Iceberg is its support for ACID transactions, which ensures data integrity by enabling atomicity, consistency, isolation, and durability in data processing operations. Additionally, Apache Iceberg's Time Travel capability allows users to query historical versions of data, providing valuable insights into data changes over time. The format also supports dynamic schema evolution and partitioning adjustments without downtime, which simplifies the management of data as it grows and changes. Apache Iceberg also includes a host of other advanced features that enhance data usability and management. These include snapshot isolation for concurrent data access, upserts and deletes within tables, and handling large-scale metadata efficiently. To help you harness the full potential of Apache Iceberg, our latest blog series gathers a collection of straightforward, hands-on tutorials. These guides are designed to provide you with practical experience in working with Apache Iceberg, from setting up your first table to executing complex data operations. Whether you're a data scientist, engineer, or analyst, these tutorials offer a valuable opportunity to enhance your data handling skills and leverage Apache Iceberg’s powerful features in your projects. Self-Contained Exercises These are completely self-contained and can be done from your laptop; all infrastructure is spun up as Docker containers. End-to-End Basic Data Engineering Tutorial (Spark, Dremio, Superset) From Postgres to Dashboards with Dremio and Apache Iceberg From SQLServer to Dashboards with Dremio and Apache Iceberg From MongoDB to Dashboards with Dremio and Apache Iceberg Intro to Dremio, Nessie, and Apache Iceberg on Your Laptop Using Flink with Apache Iceberg and Nessie Getting Started with Project Nessie, Apache Iceberg, and Apache Spark Using Docker End-to-End Basic Data Engineering Tutorial (Spark, Dremio, Superset) End-to-End Basic Data Engineering Tutorial (Spark, Dremio, Superset) End-to-End Basic Data Engineering Tutorial (Spark, Dremio, Superset) From Postgres to Dashboards with Dremio and Apache Iceberg From Postgres to Dashboards with Dremio and Apache Iceberg From Postgres to Dashboards with Dremio and Apache Iceberg From SQLServer to Dashboards with Dremio and Apache Iceberg From SQLServer to Dashboards with Dremio and Apache Iceberg From SQLServer to Dashboards with Dremio and Apache Iceberg From MongoDB to Dashboards with Dremio and Apache Iceberg From MongoDB to Dashboards with Dremio and Apache Iceberg From MongoDB to Dashboards with Dremio and Apache Iceberg Intro to Dremio, Nessie, and Apache Iceberg on Your Laptop Intro to Dremio, Nessie, and Apache Iceberg on Your Laptop Intro to Dremio, Nessie, and Apache Iceberg on Your Laptop Using Flink with Apache Iceberg and Nessie Using Flink with Apache Iceberg and Nessie Using Flink with Apache Iceberg and Nessie Getting Started with Project Nessie, Apache Iceberg, and Apache Spark Using Docker Getting Started with Project Nessie, Apache Iceberg, and Apache Spark Using Docker Getting Started with Project Nessie, Apache Iceberg, and Apache Spark Using Docker Require Online Services These are tutorials that will require signing up for services like Upsolver, Dremio, AWS, and others to complete. How to Convert JSON Files Into an Apache Iceberg Table with Dremio How to Convert CSV Files into an Apache Iceberg table with Dremio Run Graph Queries on Apache Iceberg Tables with Dremio & Puppygraph BI Dashboards with Apache Iceberg Using AWS Glue and Apache Superset Streaming and Batch Data Lakehouses with Apache Iceberg, Dremio and Upsolver Git for Data with Dremio’s Lakehouse Catalog: Easily Ensure Data Quality in Your Data Lakehouse How to Create a Lakehouse with Airbyte, S3, Apache Iceberg, and Dremio How to Convert JSON Files Into an Apache Iceberg Table with Dremio How to Convert JSON Files Into an Apache Iceberg Table with Dremio How to Convert JSON Files Into an Apache Iceberg Table with Dremio How to Convert CSV Files into an Apache Iceberg table with Dremio How to Convert CSV Files into an Apache Iceberg table with Dremio How to Convert CSV Files into an Apache Iceberg table with Dremio Run Graph Queries on Apache Iceberg Tables with Dremio & Puppygraph Run Graph Queries on Apache Iceberg Tables with Dremio & Puppygraph Run Graph Queries on Apache Iceberg Tables with Dremio & Puppygraph BI Dashboards with Apache Iceberg Using AWS Glue and Apache Superset BI Dashboards with Apache Iceberg Using AWS Glue and Apache Superset BI Dashboards with Apache Iceberg Using AWS Glue and Apache Superset Streaming and Batch Data Lakehouses with Apache Iceberg, Dremio and Upsolver Streaming and Batch Data Lakehouses with Apache Iceberg, Dremio and Upsolver Streaming and Batch Data Lakehouses with Apache Iceberg, Dremio and Upsolver Git for Data with Dremio’s Lakehouse Catalog: Easily Ensure Data Quality in Your Data Lakehouse Git for Data with Dremio’s Lakehouse Catalog: Easily Ensure Data Quality in Your Data Lakehouse Git for Data with Dremio’s Lakehouse Catalog: Easily Ensure Data Quality in Your Data Lakehouse How to Create a Lakehouse with Airbyte, S3, Apache Iceberg, and Dremio How to Create a Lakehouse with Airbyte, S3, Apache Iceberg, and Dremio How to Create a Lakehouse with Airbyte, S3, Apache Iceberg, and Dremio Video Playlists: Hands-On With Apache Iceberg Apache Iceberg Lakehouse Engineering 101
In summary, Apache Iceberg is more than just a data storage format; it is a comprehensive solution that adapts to the complexities of modern data environments, ensuring robust data integrity, flexibility, and scalability. Hands-On With Apache Iceberg Hands-On With Apache Iceberg Hands-On With Apache Iceberg Hands-On With Apache Iceberg Apache Iceberg Lakehouse Engineering 101 In summary, Apache Iceberg is more than just a data storage format; it is a comprehensive solution that adapts to the complexities of modern data environments, ensuring robust data integrity, flexibility, and scalability. Apache Iceberg Lakehouse Engineering 101 Apache Iceberg Lakehouse Engineering 101 Apache Iceberg Lakehouse Engineering 101 In summary, Apache Iceberg is more than just a data storage format; it is a comprehensive solution that adapts to the complexities of modern data environments, ensuring robust data integrity, flexibility, and scalability.

16 Guides to Get You Started with Apache Iceberg

Introduction: Large enterprises, which have more than one business unit, usually have more than one data platform environment. In fact, companies trying to be agile to make quick decisions, all have business unit-specific data platforms. With the computational advancements and rapid growth in data analytics and science, data integration tasks for data platforms are increasing exponentially as they need to support transactional as well as batch processing jobs. Here, the challenge is choosing the orchestration tool, as data integrations have to work in conjunction with different platforms. Though there are several tools in the market, in the article, we will cover how Apache airflow became the most trusted tool for the orchestration industry and its advantages. ETL: Traditional ETL (Extract, Transform, and loading) tools like Informatica, Ab Initio, and Talend all come with their own orchestration process. But as they are proprietary tools, it's expensive as well. Cloud service providers Azure, AWS (Amazon Web Services), and Google Cloud are all trying cash in the orchestration space so Azure has Synapse and AWS has Glue, Step function, etc. Basics: Orchestration is a simple or complex series of workflows defining a process. Workflows depend on one another to perform a business function or a series of business func‐tions calling other services. Then what is the ? process A process is a business function or set of business functions achieved by executing one or more tasks. Then what is the ? task A task is a step in a process. A task may call a service to perform a particular data-related function. Then what is the ? Service A service is software like Azure Synapse Analytics, Python data frame, or AWS DynamoDB that performs tasks. Common Problems in the Legacy Tools: The most common problem in the orchestration tools is their ability to successfully implement and maintain processes across the organization. Engineers and managers want to choose a tool that addresses the below issues: Dependency management: Lack of dependency management results in maintaining process-to-process and task-to-tasks dependency checks within the codebase along with business logic. This increases development time based on the number of processes or tasks involved in a workflow. Task modification to existing processes: Workflow is defined as each step within the code. Embedding the workflow in the code requires manual effort to fully review the code to make changes to one or more tasks resulting in additional time involved in project enhancements and testing. Troubleshooting: Troubleshooting a particular process should not require the need to understand the dependencies, as it results in additional support time. Restart: It should be fairly easy to restart the failed task or trigger an ad-hoc task. Orchestration should not warrant the need to rerun the entire process if a particular task within the process failed. User-friendly interface to track the progress: Orchestration tools should provide best in class user interface with the details to visualize the tasks involved within the processes in one place. Engineers investigating the issue should have a UI to understand the flow of a particular process. UI should display the current process in execution, the time it's taking, etc. Apache Airflow Airflow is a platform to programmatically author, schedule, and monitor workflows geared at data pipelining. Airflow is used to author workflows as DAGs of tasks. The Airflow scheduler executes tasks on an array of workers while following the specified dependencies. The rich user interface makes it easy to visualize pipelines running in production, monitor progress, and troubleshoot issues when needed. Code Development: Authoring a data pipeline using existing tasks involves writing a Python module and deploying it to the Airflow installation. Adding a new kind of task to an airflow installation involves writing and deploying Python code in packages called plugins. This aligns with the primary coding skill set of Data Engineers and scientists, which is Python. Benefits At a high level, process dependency management is easy to develop and maintain via user interface (UI) and support in Airflow, which helps immensely to save time for developers and supporting resources. DAG creation is all in Python scripting language which is the primary coding skillset for Data Engineers. Changes within Airflow can be captured using the GIT repository enabling version controls and change management practices. Airflow has good documentation and an active internal community for answering questions. What are the skillsets required to develop and maintain Airflow? Python is a major skill required to use Airflow. Initial learnings are required to set up Airflow and understand the architecture so it's evident for future maintenance. Can Engineers visualize the Jobs? Yes, Airflow has a Web interface to start and stop processes and also to monitor the current status of the processes. Does Airflow support any integration with Data Quality checkers? DataDog can be integrated with Airflow. How do we make ensure Airflow’s availability? Airflow workers and Web servers can be load balanced. Airflow Scheduler is a single point of failure. How does Airflow scale? Airflow Web server can be scaled based on the number of requests. Airflow workers can be scaled up and down based on requirements.

Apache Airflow And Its Contribution to Enterprise Data Integration

, a powerful open-source search engine, is known for its versatility and flexibility when it comes to handling large volumes of text data. In our continuous effort to improve the user experience and bring advanced search capabilities to a broader audience, we are excited to announce the integration of Manticore Search in , a leading data exploration and visualization platform. Manticore Search Apache Superset This integration is available since Manticore Search version 6.2.0 and allows users to harness the full potential of Manticore Search using the default MySQL connector in Apache Superset, simplifying data analysis and empowering users to create stunning visualizations without the need for extensive technical knowledge. In this article, we’ll provide a step-by-step tutorial that will guide you through connecting Manticore to Apache Superset and adding a chart. Step-by-step tutorial Step 1: Install and Configure Manticore Search Prior to linking with , ensure that Manticore Search has been properly set up and installed on your server. Consult the official Manticore Search guidelines for assistance with . Manticore Search Apache Superset installation and configuration procedures Step 2: Set up Apache Superset To set up Apache Superset, follow the official . Apache Superset installation guide Step 3: Prepare Manticore Search data to visualize In the next step, you need to prepare and create tables in Manticore Search, then load them with data that will help us generate charts and visualize them using Apache Superset. Connect to Manticore using a MySQL client and execute the following queries: Here is the structure we will use: CREATE TABLE product_sales (
id bigint,
product_category text,
sale_date timestamp,
sales_amount float
) Now, let’s add some data to populate the table: INSERT INTO product_sales (product_category, sale_date, sales_amount)
VALUES
('Electronics', 1630454400, 1500),
('Fashion', 1630540800, 1200),
('Home & Garden', 1630627200, 900),
('Electronics', 1630713600, 1800),
('Fashion', 1630800000, 1100),
('Home & Garden', 1630886400, 1000); Step 4: Connecting Manticore Search to Apache Superset Log in to your dashboard and click on “+” in the top navigation bar and choose “Database Connections”. Apache Superset Choose “+ DATABASE” to add a new database. Select “ ” as the database type. MySQL In the configuration page, provide the following details: Database Name: A name for the database (e.g., “Manticore”). Fill up “Host”, “Port”, “Username” and “Display Name” Click on “Connect” to verify the connection and save. Step 5: Creating a Chart and Dashboard In the Apache Superset dashboard, click on “Dashboards” in the top navigation bar and select “New dashboard.” Open the dashboard and click on the “+” icon to add a new chart. Choose the dataset with Manticore Search that connected using the MySQL connector. Select the type of chart you want to create (e.g., bar chart, line chart, pie chart) from the list of available visualization options. For example, we’ll choose a chart. BAR CHART Step 6: Configuring the Chart Use Apache Superset’s query builder or write an SQL query to fetch the data you want to visualize from your Manticore Search database Customize the chart’s appearance, labels, and other settings as needed. Click on “UPDATE CHART” to update the chart. Add name of the chart then click “Save” to permanently save the it in Dashboard. Step 7: Reviewing Your Chart’s Results Here is a sample of the results obtained using the provided data in the newly configured chart. Conclusion The integration of Manticore Search in Apache Superset marks a significant milestone in our ongoing commitment to provide users with advanced search capabilities and seamless data visualization experiences. By following this step-by-step tutorial, you can effortlessly connect Manticore Search to Apache Superset using the default MySQL connector and create engaging, insightful charts that showcase the true value of your data. Don’t miss out on the fantastic opportunities that this powerful combination of Manticore Search and Apache Superset offers. Start exploring your data and unlocking its full potential today! Also published here.

Integrating Manticore Search with Apache Superset 

Recently, we took over a data integration project where the upstream data was delivered to Kafka. Initially, we chose the SpringBoot+Flink approach to process the incoming data (referred to as Solution 1 below). However, after facing efficiency issues in the production environment, we turned our attention to the open-source project SeaTunnel. We discovered that SeaTunnel supports Kafka as a source and started exploring and testing it. In our development environment, we experimented with over 5 million records and achieved an efficiency of around 10,000 records per second. Consequently, we abandoned Solution 1 and adopted SeaTunnel for integrated data processing (referred to as Solution 2 below). During our exploration of SeaTunnel, we identified two methods. Method 2, compared to Method 1, offers a more comprehensive solution that handles various scenarios without concerns about data misalignment caused by unexpected characters within field values. Version Information: SeaTunnel: apache-seatunnel-2.3.2-SNAPSHOT Engines: Flink 1.16.2 Zeta: Consistent with SeaTunnel Comparison Based on Solution 2, two derived methods were developed: After extensive exploration and validation in our production environment, we recommend using Method 2 of Solution 2. Now, let's dive into the main content and explore the two methods in Solution 2, allowing you to experience the magic of SeaTunnel. Solution 1: Kafka Complex JSON Parsing with Spring Boot + Flink There are many online examples available for this approach, so we won't go into much detail here. Solution 2: Kafka Complex JSON Parsing with SeaTunnel Before we proceed, let's take a look at a sample JSON data delivered to Kafka by the upstream party (part of the data has been redacted): "type": "D***",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
    "data": {
        "LSH": "187eb13****l0214723",
        "NSRSBH": "9134****XERG56",
        "QMYC": "01*****135468",
        "QMZ": "1Zr*****UYGy%2F5bOWtrh",
        "QM_SJ": "2023-05-05 16:42:10.000000",
        "YX_BZ": "Y",
        "ZGHQ_BZ": "Y",
        "ZGHQ_SJ": "2023-06-26 16:57:17.000000",
        "SKSSQ": 202304,
        "SWJG_DM": "",
        "SWRY_DM": "00",
        "CZSJ": "2023-05-05 16:42:10.000000",
        "YNSRSBH": "9134****XERG56",
        "SJTBSJ": "2023-06-26 19:29:59.0,00",
        "SJCZBS": "I"
    },
    "beforeData": null,
    "headers": {
        "operation": "INSERT",
        "changeSequence": "12440977",
        "timestamp": "2023-06-26T19:29:59.673000",
        "streamPosition": "00****3.16",
        "transactionId": "000***0006B0002",
        "changeMask": "0FFF***FF",
        "columnMask": "0F***FFFF",
        "transactionEventCounter": 1,
        "transactionLastEvent": false
    }
} Method 1: Parsing without UDF Functions Issue: Field values may contain delimiters, such as ','. This can lead to data misalignment during storage. This method primarily utilizes various transformation plugins available in the official transform-v2. It includes the , , and implementations. Replace Split SQL SeaTunnel Script: (ybjc_qrqm.conf) job.mode = "STREAMING"
job.name = "kafka2mysql_ybjc"
execution.checkpoint.interval = 60000

source {
  Kafka {
    result_table_name = "DZFP_***_QRQM1"
    topic = "DZFP_***_QRQM"
    bootstrap.servers = "centos1:19092,centos2:19092,centos3:19092"
    schema = {
      fields {
        message =  {
            data = {
                LSH = "string",
                NSRSBH =  "string",
                QMYC =  "string",
                QMZ =  "string",
                QM_SJ =  "string",
                YX_BZ =  "string",
                ZGHQ_BZ =  "string",
                ZGHQ_SJ =  "string",
                SKSSQ =  "string",
                SWJG_DM = "string",
                SWRY_DM = "string",
                CZSJ = "string",
                YNSRSBH = "string",
                SJTBSJ = "string",
                SJCZBS = "string"
            }
        }
      }  
    }
	  start_mode = "earliest"
    kafka.config = {
      auto.offset.reset = "earliest"
      enable.auto.commit = "true"
      max.partition.fetch.bytes = "5242880"
      session.timeout.ms = "30000"
      max.poll.records = "100000"
    }
  }
}

transform {
 Replace {
    source_table_name = "DZFP_***_QRQM1"
    result_table_name = "DZFP_***_QRQM2"
    replace_field = "message"
    pattern = "[["
    replacement = ""
  }
  Replace {
    source_table_name = "DZFP_***_QRQM2"
    result_table_name = "DZFP_***_QRQM3"
    replace_field = "message"
    pattern = "]]"
    replacement = ""
  }
  
  Split {
    source_table_name = "DZFP_***_QRQM3"
    result_table_name = "DZFP_***_QRQM4"
    separator = ","
    split_field = "message"
    output_fields = [zwf1,zwf2,zwf3,zwf4,zwf5,nsrsbh,qmyc,qmz,qm_sj,yx_bz,zghq_bz,zghq_sj,skssq,swjg_dm ,swry_dm ,czsj ,ynsrsbh ,sjtbsj ,sjczbs]
  }
  
  sql{
   source_table_name = "DZFP_***_QRQM4"
   query = "select replace(zwf5 ,'fields=[','') as lsh,nsrsbh,trim(qmyc) as qmyc,qmz,qm_sj,yx_bz, zghq_bz,zghq_sj,skssq,swjg_dm ,swry_dm ,czsj ,ynsrsbh ,sjtbsj ,replace(sjczbs,']}]}','') as sjczbs  from DZFP_DZDZ_QRPT_YWRZ_QRQM4 where skssq <> ' null'"
   result_table_name = "DZFP_***_QRQM5"
  }
  
}

sink {
	Console {
		source_table_name = "DZFP_***_QRQM5"
	}
	
    jdbc {
	    source_table_name = "DZFP_***_QRQM5"
        url = "jdbc:mysql://localhost:3306/dbname?serverTimezone=GMT%2b8"
        driver = "com.mysql.cj.jdbc.Driver"
        user = "user"
        password = "pwd"
        batch_size = 200000
        database = "dbname"
        table = "tablename"
        generate_sink_sql = true
        primary_keys = ["nsrsbh","skssq"]
    } 
} Data can be successfully written: Kafka Source Data: Target Data in TiDB: However, when we simulate sending data to Kafka, specifically setting a ',' as part of the 'SJTBSJ' field: Original value: Modified value: 2023-06-26 19:29:59.000 2023-06-26 19:29:59.0,00 Command to produce data to the topic: kafka-console-producer.sh --topic DZFP_***_QRQM --broker-list centos1:19092,centos2:19092,centos3:19092 Data sent: "type": "D***",
"headers": null,
"messageSchemaId": null,
"messageSchema": null,
"message": {
    "data": {
        "LSH": "187eb13****l0214723",
        "NSRSBH": "9134****XERG56",
        "QMYC": "01*****135468",
        "QMZ": "1Zr*****UYGy%2F5bOWtrh",
        "QM_SJ": "2023-05-05 16:42:10.000000",
        "YX_BZ": "Y",
        "ZGHQ_BZ": "Y",
        "ZGHQ_SJ": "2023-06-26 16:57:17.000000",
        "SKSSQ": 202304,
        "SWJG_DM": "",
        "SWRY_DM": "00",
        "CZSJ": "2023-05-05 16:42:10.000000",
        "YNSRSBH": "9134****XERG56",
        "SJTBSJ": "2023-06-26 19:29:59.0,00",
        "SJCZBS": "I"
    },
    "beforeData": null,
    "headers": {
        "operation": "INSERT",
        "changeSequence": "12440977",
        "timestamp": "2023-06-26T19:29:59.673000",
        "streamPosition": "00****3.16",
        "transactionId": "000***0006B0002",
        "changeMask": "0FFF***FF",
        "columnMask": "0F***FFFF",
        "transactionEventCounter": 1,
        "transactionLastEvent": false
    }
} After writing the data, we notice that the data is misaligned: Conclusion: This issue can still occur in production environments when fields contain delimiters like commas in addresses or remarks. Therefore, this approach is not recommended as it poses significant risks to the data. It can be used for handling simple data as an alternative approach. Method 2: Parsing with UDF Functions This method extends SeaTunnel's capabilities using UDF functions ( ), enabling nested JSON source data parsing from Kafka. It simplifies the ST script configuration significantly. https://seatunnel.apache.org/docs/2.3.2/transform-v2/sql-udf ST Script: (ybjc_qrqm_yh.conf) execution.parallelism = 5
job.mode = "STREAMING"
job.name = "kafka2mysql_ybjc_yh"
execution.checkpoint.interval = 60000
}

source {
    Kafka {
        result_table_name = "DZFP_***_QRQM1"
        topic = "DZFP_***_QRQM"
        bootstrap.servers = "centos1:19092,centos2:19092,centos3:19092"
        schema = {
        fields {
             message =  {
                data = "map<string,string>"
             }
        }  
        }
        start_mode = "earliest"
        kafka.config = {
        auto.offset.reset = "earliest"
        enable.auto.commit = "true"
        max.partition.fetch.bytes = "5242880"
        session.timeout.ms = "30000"
        max.poll.records = "100000"
        }
    }
}
transform {
    sql{
        source_table_name = "DZFP_***_QRQM1"
        result_table_name = "DZFP_***_QRQM2"
        query = "select qdmx(message,'lsh') as lsh,qdmx(message,'nsrsbh') as nsrsbh,qdmx(message,'qmyc') as qmyc,qdmx(message,'qmz') as qmz,qdmx(message,'qm_sj') as qm_sj,qdmx(message,'yx_bz') as yx_bz,qdmx(message,'zghq_bz') as zghq_bz,qdmx(message,'zghq_sj') as zghq_sj,qdmx(message,'skssq') as skssq,qdmx(message,'swjg_dm') as swjg_dm,qdmx(message,'swry_dm') as swry_dm,qdmx(message,'czsj') as czsj,qdmx(message,'ynsrsbh') as ynsrsbh, qdmx(message,'sjtbsj') as sjtbsj,qdmx(message,'sjczbs') as sjczbs  from  DZFP_DZDZ_QRPT_YWRZ_QRQM1"
    }
}

    
sink {
    Console {
        source_table_name = "DZFP_***_QRQM2"
    }
    jdbc {
        source_table_name = "DZFP_***_QRQM2"
        url = "jdbc:mysql://localhost:3306/dbname?serverTimezone=GMT%2b8"
        driver = "com.mysql.cj.jdbc.Driver"
        user = "user"
        password = "pwd"
        batch_size = 200000
        database = "dbname"
        table = "tablename"
        generate_sink_sql = true
        primary_keys = ["nsrsbh","skssq"]
    } 
} Executing the script and checking the results, we can see that the data is not misaligned and remains in the original field ('sjtbsj'): The specific UDF function is implemented as follows. Maven dependencies are as follows: <dependency>
    <groupId>org.apache.seatunnel</groupId>
    <artifactId>seatunnel-transforms-v2</artifactId>
    <version>2.3.2</version>
    <scope>provided</scope>
  </dependency>

  <dependency>
    <groupId>org.apache.seatunnel</groupId>
    <artifactId>seatunnel-api</artifactId>
    <version>2.3.2</version>
  </dependency>

  <dependency>
    <groupId>cn.hutool</groupId>
    <artifactId>hutool-all</artifactId>
    <version>5.8.20</version>
  </dependency>

  <dependency>
    <groupId>com.google.auto.service</groupId>
    <artifactId>auto-service-annotations</artifactId>
    <version>1.1.1</version>
    <optional>true</optional>
    <scope>compile</scope>
  </dependency>

  <dependency>
    <groupId>com.google.auto.service</groupId>
    <artifactId>auto-service</artifactId>
    <version>1.1.1</version>
    <optional>true</optional>
    <scope>compile</scope>
  </dependency>
</dependencies> UDF-specific implementation in Java: import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.google.auto.service.AutoService;
import org.apache.seatunnel.api.table.type.BasicType;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.transform.sql.zeta.ZetaUDF;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

@AutoService(ZetaUDF.class)
public class QdmxUDF implements ZetaUDF {

    @Override
    public String functionName() {
        return "QDMX";
    }

    @Override
    public SeaTunnelDataType<?> resultType(List<SeaTunnelDataType<?>> list) {
        return BasicType.STRING_TYPE;
    }

    // list Example of parameters: (which is the data parsed by Kafka)
    //SeaTunnelRow{tableId=, kind=+I, fields=[{key1=value1,key2=value2,.....}]}
    @Override
    public Object evaluate(List<Object> list) {
        String str = list.get(0).toString();
        //1 Remove the prefix
        str = StrUtil.replace(str, "SeaTunnelRow{tableId=, kind=+I, fields=[{", "");
        //2 Remove the suffix
        str = StrUtil.sub(str, -3, 0);
        // 3 build Map key value
        Map<String, String> map = parseToMap(str);
        if ("null".equals(map.get(list.get(1).toString())))
            return "";
        // 4 return the value of the key
        return map.get(list.get(1).toString());
    }

    public static Map<String, String> parseToMap(String input) {
        Map<String, String> map = new HashMap<>();
        // Remove the curly brackets and remove them during the string stage.
        // input = input.replaceAll("[{}]", "");
        // Split key-value pairs
        String[] pairs = input.split(", ");

        for (String pair : pairs) {
            String[] keyValue = pair.split("=");
            if (keyValue.length == 2) {
                String key = keyValue[0].trim().toLowerCase();
                String value = keyValue[1].trim();
                map.put(key, value);
            }
        }
        return map;
    }
} Then, for packaging, use the following command: mvn -T 8 clean install -DskipTests -Dcheckstyle.skip -Dmaven.javadoc.skip=true Check the directory to see if the @AutoService annotation generates the corresponding SPI interface. META-INF/services If it is as shown below, then the packaging is successful! If not, the packaging fails, and the UDF function cannot be used. You can refer to my packaging plugin: <plugins>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-site-plugin</artifactId>
      <version>3.7</version>
      <dependencies>
        <dependency>
          <groupId>org.apache.maven.doxia</groupId>
          <artifactId>doxia-site-renderer</artifactId>
          <version>1.8</version>
        </dependency>
      </dependencies>
    </plugin>
    <plugin>
      <groupId>org.apache.maven.plugins</groupId>
      <artifactId>maven-compiler-plugin</artifactId>
      <version>3.8.1</version>
      <configuration>
        <annotationProcessorPaths>
          <path>
            <groupId>com.google.auto.service</groupId>
            <artifactId>auto-service</artifactId>
            <version>1.1.1</version>
          </path>
        </annotationProcessorPaths>
      </configuration>
    </plugin>
  </plugins>
</build> Finally, place the packaged JAR file in the $ directory. Since my UDF function depends on a third-party JAR, it should be uploaded together. If using the Zeta cluster, you need to restart the Zeta cluster for it to take effect. For other engines, it will take effect in real time. {SEATUNNEL_HOME}/lib The successful upload will look like this: Note: The "hutool-all" JAR file can be included in the project. I uploaded two for convenience. java_study In conclusion, it is recommended to to parse the nested JSON data from Kafka sources. use the extension of UDF functions Also published . here

Resolving Complex JSON in Kafka Source Using Apache SeaTunnel

Apache Pulsar has always seen itself as cloud native. It’s right there on the home page: “Apache® Pulsar™ is an open-source, distributed messaging and streaming platform built for the cloud” Pulsar certainly embodies many cloud-native principles, including the separation of compute from storage where the Pulsar broker handles the serving of messages and Apache BookKeeper handles the storage of them, and the idea of stateless components that can be scaled up and down at will like the Pulsar broker. However, it has never really delivered on the ultimate promise of the cloud: autoscaling. At least, not until today. I am excited to introduce Kubernetes Autoscaling for Apache Pulsar, or KAAP. KAAP is a Kubernetes operator that is available on It includes all the niceties that you would expect from a Kubernetes operator. Once you have the operator installed in your Kubernetes cluster, you can get a fully functional Pulsar cluster by applying a single custom resource definition (CRD) like this: OperatorHub. apiVersion: kaap.oss.datastax.com/v1alpha1
kind: PulsarCluster
metadata:
  name: pulsar
spec:
  global:
    name: pulsar
    image: 'apachepulsar/pulsar:3.0.0' Pulsar has multiple components: ZooKeeper, BookKeeper, brokers, and proxies. The operator handles configuring all of them with the CRD. Because the operator works at the cluster level, you don’t have to worry about how the components work together. The operator takes care of that for you. PulsarCluster Automatic staged upgrades Having operated a production cloud service for Pulsar longer than anyone I know of (first at and now with DataStax Astra Streaming), I know firsthand how long it takes to do a proper upgrade of a Pulsar cluster. While you could just YOLO upgrade all the components at once, the recommended and safest method is to upgrade each component one at a time, ensuring each component has successfully upgraded before moving on to the next. Kesque This is the process that we follow when upgrading the Pulsar clusters in our Astra Streaming service because the availability of our customer service is paramount, but it is very time-consuming for our production engineering team. In fact, this is the number one request they have: can you make the upgrades smarter and more automatic? Well, with KAAP, we have done just that. With a single line change in the CRD, you can trigger a staged upgrade of your Pulsar cluster. The KAAP operator will orchestrate a careful, staged upgrade of your cluster. Of course, you should monitor the metrics during the upgrade, but you can let the KAAP operator drive (no sleeping at the wheel). PulsarCluster Open source autoscaling Automatic staged upgrades are nice, but that’s not what I am most excited about with KAAP. The concept of elastic resources is a key promise of the cloud. It is why many of the AWS services have “elastic” in their name. I (actually ) did a quick search and there are at least 10 of them. ChatGPT The cloud enables you to acquire and release resources on demand, so you can use autoscaling techniques to match the number of resources your application is using to the load on it. As the load increases, you can automatically add more resources. As the load decreases, you can free up resources. Since cloud providers charge for the usage of resources, autoscaling can help you make the most efficient use of your cloud spend. I am sure we can all think of examples of cloud services that feature autoscaling behind the scenes. Most of those “elastic” AWS services work like that. However, the implementation of the autoscaling is proprietary, locked in a vendor’s private repo. Sometimes they will talk about how they have implemented autoscaling, like in this about AWS DynamoDB, or this about how Confluent Cloud makes Kafka elastic. But no one, to my knowledge, has made their autoscaling implementation open source, freely available to all users under a truly open source license. That’s exactly what we are doing with KAAP. paper blog post KAAP adds sophisticated autoscaling capabilities to your Pulsar (or , our Pulsar distribution) under an Apache 2.0 license. You can check it out in our . If you like it, remember to give it a star. Luna Streaming GitHub repository Pulsar + auto scaling: A perfect match KAAP gives you a way to add autoscaling to Pulsar when running in Kubernetes. Not some simple, Horizontal Pod Autoscaler (HPA) tacked on to a Pulsar broker deployment, but a sophisticated auto scaler that works with the existing cloud-native components of Pulsar to provide a truly elastic streaming and messaging system. In my opinion, this is a combination that can’t be beat. Stateless broker scaling Let’s take a look at the two main autoscaling dimensions of KAAP. First, we’ll look at Pulsar brokers. As you might know, Pulsar brokers are stateless, which means you can scale them up and down easily, as required. But what you might not know is that Pulsar has a built-in broker load-balancing mechanism that continuously monitors CPU, memory, and network bandwidth on each broker. Using that information and one of the configurable load-balancing algorithms, Pulsar will move topics from broker to broker to prevent a broker from being overloaded. A naive autoscaling solution would be to configure a Kubernetes Horizontal Pod Autoscaler (HPA) on the brokers and when some metric like CPU gets high, scale up another broker pod. But this might not actually be necessary because the Pulsar broker load balancer could decide to shift topics to even out the load at the same time. Now you’ve scaled up a broker pod that isn’t needed because the Pulsar load balancer has balanced things out. So now the HPA decides to scale down the new broker pod, which causes any new topics that had been created on it to get moved to an existing broker.  As you can imagine, the Pulsar load balancer and an HPA can create a thrashing mess of brokers going up and down and topics being shifted from broker to broker. KAAP avoids this problem by integrating directly with the Pulsar load balancer. KAAP scales up the brokers when the cluster-wide metrics from the Pulsar load balancer suggest that the cluster is nearing capacity, not when a single broker pod is busy. And it only scales down a broker if the usage of the whole cluster falls below a configured threshold. The KAAP operator works with the Pulsar broker load balancer, not against it. Scaling storage up and down: Remarkable! Scaling the compute (or serving) layer of Pulsar is great, but it’s not enough for a true autoscaling implementation. Sure, the number of messages (or events) that need to be processed can vary over time, but what about the number of messages that need to be stored? We’ve probably all had to deal with a backlog of messages building up because of a failure of a downstream system. As the outage drags on, the available storage on the streaming system starts to run low. This is a scenario where Pulsar and its reliance on BookKeeper shines. To add storage capacity to a Pulsar cluster, you just need to add a new BookKeeper node, affectionately called a “bookie.” Because BookKeeper storage is based on segments of topics and not whole topics, the new bookie is immediately available to relieve the pressure on storage, with any painfully rebalancing of topics or any other operational intervention. KAAP can, of course, handle this case for you. It constantly monitors the disk usage of the bookie nodes, and if available storage is getting low, it will scale up a new bookie node. This is not all that remarkable (at least for Pulsar). It’s pretty straightforward to add a new replica in Kubernetes, even if it is stateful and backed by persistent volumes. But how about when the outage is over and the backlog has cleared? Are you now stuck with an extra bookie node consuming resources that aren’t really needed anymore? Not with KAAP, you’re not. Once the BookKeeper storage drops below a configured threshold, the KAAP operator will carefully orchestrate the removal of the unneeded bookie node. It does this in a very safe way, ensuring that no messages are lost and that the required replication factor is maintained at all times. For example, if you have configured Pulsar to keep three copies of each message (write and ack quorum both equal three), KAAP interacts with the BookKeeper to copy messages from the bookie that is being scaled down to other bookies in order to ensure there are at least three copies of the message available. Once that has been completed successfully, it will proceed with removing the unneeded bookie. With KAAP you get automatic scaling of the storage in your Pulsar cluster both up and down so that you can optimize the storage usage in your cluster and not get stuck with idle capacity after an unfortunate production incident. I don’t know about you, but I think that’s pretty remarkable. Zone awareness and migration tools KAAP can do staged upgrades and a smart scaling of your Pulsar cluster. But there’s more. To operate a highly available cluster in a cloud provider, it’s important to take availability zones (AZ) into consideration. If you don’t spread your components, especially BookKeeper, across availability zones, you won’t be able to survive an AZ failure and provide multiple nines of availability. Luckily Pulsar has great built-in capabilities like rack awareness to support high-availability deployments. The tricky part is that to set it properly you need to both configure Kubernetes correctly with zone awareness and also configure Pulsar. The KAAP operator has you covered by introducing the concept of resource sets, which enable you to group components and give them rack awareness. The KAAP operator will automatically apply both the Kubernetes and Pulsar configurations based on your declarative configuration of resource sets. Resource sets are flexible, allowing you to support a variety of Pulsar deployment options. And what if you are already running your Pulsar using a Helm chart or maybe just some Kubernetes manifest magic? KAAP has a migration tool to help you out. You can point the migration tool at your existing Kubernetes Pulsar deployment and it will automatically generate a matching CRD configuration that you can use to have the KAAP operator take over operating your cluster for you. The KAAP Stack The KAAP operator has lots of great features, turbocharging your regular Pulsar cluster into a well-oiled, highly available, autoscaling machine. But as someone who has operated production Pulsar clusters for a long time, I know that there are lots of other considerations for creating a production Pulsar cluster, such as TLS certificate management, authentication, and monitoring. That’s why we have included what we call the KAAP stack with the operator. It’s an umbrella Helm chart that installs the KAAP operator along with critical production tools, including: Cert Manager Keycloak Prometheus Stack (Grafana) Pulsar Grafana dashboards These are must-have tools when running our production Pulsar clusters and we didn’t want to leave you high and dry, so we pulled them all together and integrated them into one convenient package. Why use KAAP? So you’ve heard about all the great features of Kubernetes Autoscaling for Apache Pulsar. You can create an entire Pulsar cluster with a single CRD. You can put your upgrades on autopilot and let the KAAP operator perform safe, staged upgrades. You can automatically scale Pulsar brokers up and down based on the Pulsar broker load balancer saying the brokers are getting overloaded, and you can scale BookKeeper nodes up and down (!) safely based on the storage requirements of your clusters. You can easily configure your cluster for availability zone awareness for high availability. And it even includes a migration tool so you can easily move from your old Helm-based deployment to a turbo-charged, KAAP operator-based one. So—lots of great features, but what are the benefits of KAAP? I can think of several: Easily configure and operate highly available Pulsar clusters running in Kubernetes leading to less effort for you and your production teams Dramatically simplify scaling Pulsar to match changing demands by autoscaling the cluster resources to match the demand Reduce the total cost of ownership of a Pulsar cluster by eliminating provisioning to peak loads or over provisioning as result of production incidents Avoid vendor lock-in by using all open source technologies In my opinion, releasing KAAP is a truly innovative moment in the streaming and messaging space. No other open-source project combines the streaming and messaging power of Apache Pulsar with the ultimate promise of cloud computing: fully elastic, autoscaling. I’m looking forward to you trying it out. Hop into GitHub Discussions in our repo and let us know what you think! Find out more If you want to take a technical deep dive into KAAP, take a look at . You can find the complete documentation for KAAP . And is the GitHub repository. this blog post here here By Chris Bartholomew, DataStax This story was distributed by Datastax under HackerNoon’s Brand As An Author Program. Learn more about the program here: https://business.hackernoon.com/brand-as-author

Stories, News & Mentions about APACHE

Categories

Trending Topics

Stories, News & Mentions about APACHE

Apache’s Stories on HackerNoon

Apache's News & Mentions

Categories

Trending Topics