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

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

Fulfilling the Cloud Native Promise: Autoscaling for Apache Pulsar Is Here!

Streaming que cambia el juego: Apache Pulsar presenta el escalado automático

Streaming qui change la donne : Apache Pulsar présente l'autoscaling

गेम-चेंजिंग स्ट्रीमिंग: अपाचे पल्सर ने ऑटोस्केलिंग की शुरुआत की

革新的なストリーミング: Apache Pulsar が自動スケーリングを導入

Streaming revolucionário: Apache Pulsar apresenta escalonamento automático

Truyền trực tuyến thay đổi trò chơi: Apache Pulsar giới thiệu Autoscaling

改变游戏规则的流媒体：Apache Pulsar 推出自动缩放功能

Schemas are an essential part of any They’re metadata that define the shape of the data and the properties’ names and data types. Schemas are helpful in two ways. First, they provide a fixed blueprint for the data format, which can prevent badly formed data from being used within the context of the schema. Second, schemas let users understand how to parse the data and what to expect from its properties. data platform. is an open source, distributed publish-subscribe (pub-sub) messaging system that enables server-to-server data transportation . Messages are sent via topics, with the producer putting messages onto these topics to be read by the consumer. A user can define a schema for a topic stored in Pulsar’s schema registry. When messages are added to the topic, the Pulsar broker checks that the message conforms to the schema and ensures only valid messages are sent. The schema acts as a contract between the producer and consumers, enabling both parties to know the exact format of the data. Apache Pulsar using schema Over time, as applications evolve, it may be that the data produced by the application changes too. However, schema changes can affect the downstream consumers of the data that expect data in a specific format. Without a way to manage schemas between producers and consumers, it’s difficult to make changes to the data that is written in the messages or events without risking breaking downstream applications. To avoid this kind of issue, the schema of the messages must also evolve as new properties are added and allow consumers to understand data in both the old and new formats. This concept is known as schema evolution and Pulsar supports it. This article discusses why schemas evolve before diving into how Pulsar implements and supports schema evolution. Why Schemas Evolve Schemas provide context about raw data. They often describe a particular entity in a system, encompassing all the properties of that entity. For example, you may have an application that signs up users. It stores user details such as names, email addresses and ages. There would be a user schema that describes the underlying data and provides context, such as the name of the field and the type of data in it. The schema may look as follows: Now, say you wanted to expand the data captured to include address data and support direct mailing to the users. You would then need to expand the schema to include the new fields for capturing addresses, such as the first line of the address, city and ZIP code. After including these new fields, the schema will look like this: This is a simple form of schema evolution, as the original fields haven’t changed, and only new fields have been added. In most cases, this shouldn’t be a breaking change for downstream consumers, as consumers can continue as if the new fields were absent. The consumers would just need to be updated to consume and use the new properties. However, sometimes existing fields need to be amended to support new functionality. For example, say that users were uncomfortable giving an exact age, and instead, you change the application to capture age ranges like 18-24, 25-39,40-49 and 60+. The age column would need its data type amended from integer to string. This is a more complex schema evolution, as this could break downstream consumers who are processing the age property and expecting it to be a number or parsing the number using a strictly typed language like Java. They could also perform numeric calculations on the property, which would no longer work in its new format. To overcome this challenge, data platforms can support schema evolution to handle scenarios like this. Pulsar recognizes the importance of schema to data processing; in fact, it treats it like a first-class citizen by including built-in schema evolution support. Let’s look at how Pulsar does this. How Pulsar Supports Schema Evolution Pulsar defines schemas in a data structure called `SchemaInfo`. This is a Pulsar-specific data structure, which is itself a schema, that describes the schema of messages transported via Pulsar. Each `SchemaInfo` belongs to a topic (message channel) and describes the messages to be produced and consumed using that topic. Each `SchemaInfo` has a type that details the type of schema being used. This can be anything from an integer, a string, or a complex schema such as Avro or Protobuf. To support , Pulsar uses schema compatibility checks to compare an incoming schema to a topic’s known schema(s). Schema compatibility checks occur when a producer or consumer attempts to connect to a particular topic. The strategy is chosen while configuring the broker, with the value `schemaCompatibilityStrategy`. The goal is to check the `SchemaInfo` provided by the connecting client with existing `SchemaInfo` to see if they are compatible. schema evolution Pulsar supports eight different types of , which you can set based on the requirements of the change. These strategies also come with rules outlining what changes can be made and pointers around which clients to update first. (Check out the documentation linked above for a deeper explanation of each compatibility strategy). schema compatibility strategies Returning to the earlier examples, let’s implement the schema changes using Pulsar’s compatibility strategy. First, start with the initial user schema (without the address). This will be V1 of your schema. So, when you implement a Pulsar producer or consumer for the first time, the `SchemaInfo` for this version will be stored, and the producer and consumer will work as expected. Next, you want to add the new address fields to your user schema. The first step is to consult the schema compatibility strategies and determine which one is best for this change. Using the “changes allowed” column in the documentation, you’re looking for any strategy that allows the addition of new fields. This gives you BACKWARD, BACKWARD_TRANSITIVE, FORWARD, and FORWARD_TRANSITIVE. BACKWARD should be used when there’s no guarantee that consumers using an older version can understand the new schema. FORWARD is used when the consumers on the latest schema version might not be able to read data within older versions. If you want to update all of the consumers first to use the new schema, use a BACKWARD strategy. Otherwise, FORWARD is best. Looking at the bigger picture, Pulsar refers to the entire act of evolving a topic’s schema as . It’s the combination of a producer or consumer providing `SchemaInfo`, the chosen compatibility strategy for the topic, and the broker deciding what to do. schema verification Conclusion It’s rare for schemas to stay the same forever. As new features are introduced into applications, schemas often have to evolve to support these features. However, keeping the data's producers and consumers in sync can often be a challenge when schemas are modified. Pulsar’s built-in schema evolution concepts help deal with these changes. Using schema compatibility strategies, it can define the rules of how different compatible versions of a schema are. Pulsar uses this in conjunction with a schema verification process that then uses these rules to determine which schemas can be used by a consumer when connecting to a particular topic. Also published here.

How to Manage Schema Change with Apache Pulsar

Cómo administrar el cambio de esquema con Apache Pulsar

Comment gérer le changement de schéma avec Apache Pulsar

अपाचे पल्सर के साथ स्कीमा चेंज को कैसे प्रबंधित करें

Apache Pulsar を使用してスキーマ変更を管理する方法

Como gerenciar a mudança de esquema com o Apache Pulsar

Cách quản lý thay đổi lược đồ với Apache Pulsar

如何使用 Apache Pulsar 管理模式更改

Apache Cassandra is becoming the best database for handling JSON documents. If you’re a Cassandra developer who finds that statement provocative, read on. In a previous article, to mold Cassandra into a developer experience more idiomatic to the way developers think, thus improving developer productivity while preserving reasonable database performance and scale. It’s a great hypothesis and one that needs to be tested in the context of a particular developer idiom and developer community. we discussed using data APIs and data modeling , an object data mapping library usually used with the MongoDB driver, is an open source project with a significant community of JavaScript developers around it. At the , the open source API data gateway designed to work with Cassandra, we’ve partnered with Mongoose, and we’re working on an upcoming that will be released together with a that will work through that JSON API to connect to Cassandra. This creates an end-to-end stack for Mongoose developers that is fully open-source. It’s game-changing for Mongoose developers, and opens an important new chapter for Cassandra. Mongoose Stargate project JSON API version of Mongoose In this article, I’ll discuss how to provide a developer-friendly JSON idiom using Cassandra together with Stargate, and how we’re working to do just that for Mongoose developers. The Goldilocks of JS Communities In October 2022, we released a new version of Stargate. With the new , individual APIs are no longer embedded in the core Stargate coordinator code, but instead separated out into individual services. This improves Stargate’s operational efficiency; individual API services can now be deployed and scaled independently. This also makes new API services easier to develop. As long as they abide by the service boundary, these services can be developed in parallel with and independent of core Stargate development work. Version 2 We then looked for a truly idiomatic experience we could deliver to developers. With 18 million developers, JavaScript is the world’s most popular programming language, and JSON is the standard way that JavaScript developers structure data. However, 18 million people is not a community; it is many communities. We needed the “Goldilocks” of JavaScript communities — big enough to be significant, but small enough to be focused. We found the right community built around Mongoose, an object data mapper library used with applications that connect to MongoDB. Mongoose has several important characteristics: It’s JavaScript-centric It enjoys broad adoption, with roughly 2 million GitHub repositories listing Mongoose as a dependency Mongoose creator ’s active leadership provides a clear focus Valeri Karpov It’s an open source project that has lacked an open source database since MongoDB’s decision to move to a shared source model with its Server Side Public License Developers don’t really interact directly with a database so much as a data model. In Stargate’s original Document API, the API handles JSON by making it look like a traditional Cassandra table. This puts a burden on JSON-oriented developers to think in terms of Cassandra data structures, and puts a burden on Cassandra’s row-oriented indexing logic because a JSON document gets spread across multiple rows. Our new JSON API departs from this data model, and instead relies on a data model we call “super shredding.” You can learn more about super shredding by watching Aaron Morton’s talk at the recent event. In short, we take advantage of Cassandra’s wide-column nature to store one document per row, knowing that a Cassandra row can handle even very large documents. We also have a set of columns in that row that are explicitly for storing standard metadata characteristics of a JSON document. Now we have something more easily indexable, as well as a means of preserving and retrieving metadata. Cassandra Forward We will then front this data model with our new JSON API, using the same specification that Mongoose uses as our guiding requirement for which calls the API needs to support. When complete, this should enable any of the more than 2 million Mongoose-dependent applications to run against open-source Cassandra or DataStax’s hosted Cassandra service, , with just a configuration change. mQuery Astra DB With Mongoose and the new JSON API, we’ll provide a fully idiomatic experience to JSON-oriented JavaScript developers, giving them the underpinning an authentic JSON data model. scale and performance of Cassandra Mongoose creator Karpov also spoke at the recent Cassandra Forward event (you can watch the replay ), demonstrating a simple e-commerce application that uses the Stargate version of Mongoose, open source Stargate and the DataStax Enterprise (DSE) version of Cassandra. You can download the working code for this application and the supporting platform pieces from . While we have enough code to run this application, we are not yet code complete. For example, we run against DSE right now because we need (SAI), which works with DSE and is planned for release in Cassandra 5.0 later this year. here GitHub storage-attached indexing Contributing Back to Cassandra Cassandra isn’t a static piece of software; it’s a vibrant and evolving open-source project. So we are also continuing a longstanding Cassandra tradition of using that emerge client-side to foster changes on the database side.o  Similarly, Stargate’s Mongoose work has prompted a set of proposals for Cassandra around global sort and advanced query filtering that will not only make Stargate’s JSON API and Mongoose client better, but will add powerful new features to Cassandra Query Language. This is a great reminder that data engineers and application developers are not two different communities, but complementary cohorts of the extended Cassandra community. features like SAI And JSON is just the first step. Essentially, what we will have done is to take the building blocks of Cassandra, Stargate and a reasonably efficient Cassandra data model and build a document database that you interact with through a JSON API. In other words, we’ve used super shredding to create a purpose-built database that better serves the community of Mongoose developers. With the modular architecture of , and the proof point of Mongoose for the idiomatic approach, we are ready to take on new developer communities that organize around a particular software development idiom. The process by which we’ve harnessed Cassandra for Mongoose is repeatable — and it’s one that we will repeat. In so doing, we dramatically expand the number of developers and use cases that Cassandra can address, which is the sort of goal worthy of an open-source project. Stargate v2 . Learn more about DataStax By Mark Stone. Mark is product manager at DataStax. He's a technology veteran with many years of experience in product management, program management and people management. Always working as part of the connective tissue between business stakeholders and technical stakeholders, Mark loves championing the developer experience in technology platforms and helping organizations meet developers where they are.

Stories, News & Mentions about APACHE

Apache’s Stories on HackerNoon

Apache's News & Mentions