A unified serverless stream processing platform to manage, secure and process all your event streams

DeltaStream

deltastream

T3Go is China’s first platform for smart travel based on the Internet of Vehicles. In this article, Trevor Zhang and Vino Yang from T3Go describe the evolution of their data lake architecture, built on cloud-native or open-source technologies, including Alibaba OSS, Apache Hudi, and Alluxio. Today, their data lake stores petabytes of data, supporting hundreds of pipelines and tens of thousands of tasks daily. It is essential for business units at T3Go, including Data Warehouse, Internet of Vehicles, Order Dispatching, Machine Learning, and self-service query analysis. In this blog, you will see how we slashed data ingestion time by half using Hudi and Alluxio. Furthermore, data analysts using Presto, Hudi, and Alluxio saw the queries speed up by 10 times. We built our data lake based on data orchestration for multiple stages of our data pipeline, including ingestion and analytics. T3Go data lake Overview Prior to the data lake, different business units within T3Go managed their own data processing solutions, utilizing different storage systems, ETL tools, and data processing frameworks. Data for each became siloed from every other unit, significantly increasing cost and complexity. Due to the rapid business expansion of T3Go, this inefficiency became our engineering bottleneck. We moved to a unified data lake solution based on Alibaba OSS, an object store similar to AWS S3, to provide a centralized location to store structured and unstructured data, following the design principles of ; all the applications access OSS storage as the source of truth, as opposed to different data silos. This architecture allows us to store the data as-is without having to first structure the data and run different types of analytics to guide better decisions, building dashboards and visualizations from big data processing, real-time analytics, and machine learning. Multi-cluster Shared-data Architecture Efficient Near Real-time Analytics Using Hudi Our business in smart travel drives the need to process and analyze data in a near real-time manner. With a traditional data warehouse, we faced the following challenges: High overhead when updating due to long-tail latency. High cost of order analysis due to the long window of a business session Reduced query accuracy due to late or ad-hoc updates Unreliability in data ingestion pipeline. Data lost in the distributed data pipeline cannot be reconciled. High latency to access data storage. As a result, we adopted Apache Hudi on top of OSS to address these issues. The following diagram outlines the architecture: Enable near real-time data ingestion and analysis With Hudi, our data lake supports multiple data sources, including Kafka, MySQL binlog, GIS, and other business logs in near real-time. As a result, more than 60% of the company’s data is stored in the data lake, and this proportion continues to increase. We are also able to speed up the data ingestion time down to a few minutes by introducing Apache Hudi into the data pipeline. Combined with big data interactive query and analysis frameworks, such as Presto and SparkSQL, real-time data analysis and insights are achieved. Enable Incremental processing pipeline With the help of Hudi, it is possible to provide incremental changes to the downstream derived table when the upstream table updates frequently. Even with a large number of interdependent tables, we can quickly run partial data updates. This also effectively avoids updating the full partitions of cold tables in the traditional Hive data warehouse. Accessing Data using Hudi as a unified format Traditional data warehouses often deploy Hadoop to store data and provide batch analysis. Kafka is used separately to distribute Hadoop data to other data processing frameworks, resulting in duplicated data. Hudi helps effectively solve this problem; we always use Spark pipelines to insert new updates into the Hudi tables, then incrementally read the update of Hudi tables. In other words, Hudi tables are used as the unified storage format to access data. Efficient Data Caching Using Alluxio In the early version of our data lake without Alluxio, data received from Kafka in real-time is processed by Spark and then written to OSS data lake using Hudi DeltaStreamer tasks. With this architecture, Spark often suffered high network latency when writing to OSS directly. Since all data is in OSS storage, OLAP queries on Hudi data may also be slow due to a lack of data locality. To address the latency issue, we deployed Alluxio as a data orchestration layer, co-located with computing engines such as Spark and Presto, and used Alluxio to accelerate read and write on the data lake as shown in the following diagram: Data in formats such as Hudi, Parquet, ORC, and JSON are stored mostly on OSS, consisting of 95% of the data. Computing engines such as Flink, Spark, Kylin, and Presto are deployed in isolated clusters, respectively. When each engine accesses OSS, Alluxio acts as a virtual distributed storage system to accelerate data, being co-located with each of the computing clusters. Specifically, here are a few applications leveraging Alluxio in the T3Go data lake. Data lake ingestion We mount the corresponding OSS path to the Alluxio file system and set Hudi’s parameter value to use the scheme in place of scheme. Spark pipelines with Hudi continuously ingest data to Alluxio. After data is written to Alluxio, it is asynchronously persisted from the Alluxio cache to the remote OSS every minute. These modifications allow Spark to write to a local Alluxio node instead of writing to remote OSS, significantly reducing the time for the data to be available in the data lake after ingestion. target-base-path alluxio:// oss:// Data analysis on the lake We use Presto as an ad-hoc query engine to analyze the Hudi tables in the lake, co-locating Alluxio workers on each Presto worker node. When Presto and Alluxio services are co-located and running, Alluxio caches the input data locally in the Presto worker, which greatly benefits Presto for subsequent retrievals. On a cache hit, Presto can read from the local Alluxio worker storage at memory speed without any additional data transfer over the network. Concurrent accesses across multiple storage systems In order to ensure the accuracy of training samples, our machine learning team often synchronizes desensitized data in production to an offline machine learning environment. During synchronization, the data flows across multiple file systems, from production OSS to an offline HDFS followed by another offline Machine Learning HDFS. This data migration process is not only inefficient but also error-prone for modelers because multiple different storages with varying configurations are involved. Alluxio helps in this specific scenario by mounting the destination storage systems under the same filesystem to be accessed by their corresponding logical paths in Alluxio namespace. By decoupling the physical storage, allows applications with different APIs to access and transfer data seamlessly. This data access layout also improves performance. Microbenchmark Overall, we observed the following improvements with Alluxio: It supports a hierarchical and transparent caching mechanism. It supports cache promote mode when reading. It supports asynchronous writing mode. It supports LRU recycling strategy. It has pin and TTL features After comparison and verification, we choose to use Spark SQL as the query engine. Our performance testing queries the Hudi table, comparing Alluxio + OSS together against OSS directly as well as HDFS. In the stress test shown above, after the data volume is greater than a certain magnitude (2400W), the query speed using Alluxio+OSS surpasses the HDFS query speed of the hybrid deployment. After the data volume is greater than 1E, the query speed starts to double. After reaching 6E data, it is up to 12 times higher than querying native OSS and 8 times higher than querying native HDFS. The improvement depends on the machine configuration. Based on our performance benchmarking, we found that the performance can be improved by over 10 times with the help of Alluxio. Furthermore, the larger the data scale, the more prominent the performance improvement. Next Step As T3Go’s data lake ecosystem expands, we will continue facing the critical scenario of compute and storage segregation. With T3Go’s growing data processing needs, our team plans to deploy Alluxio on a larger scale to accelerate our data lake storage. In addition to the deployment of Alluxio on the data lake computing engine, which currently is mainly SparkSQL, we plan to add a layer of Alluxio to the OLAP cluster using Apache Kylin and an ad_hoc cluster using Presto. The goal is to have Alluxio cover all computing scenarios, with Alluxio interconnected between each scene to improve the read and write efficiency of the data lake and the surrounding lake ecology. Conclusion As mentioned earlier, Hudi and Alluxio cover all scenarios of Hudi’s near real-time ingestion, near real-time analysis, incremental processing, and data distribution on DFS, among many others, and play the role of a powerful accelerator on data ingestion and data analysis on the lake. With Hudi and Alluxio together, our R&D engineers shortened the time for data ingestion into the lake by up to a factor of 2. Data analysts using Presto, Hudi, and Alluxio in conjunction to query data on the lake saw their queries speed up by 10 times faster. Furthermore, the larger the data scale, the more prominent the performance improvement. Alluxio is an important part of T3Go’s plan to become a leading enterprise data lake in China. We look forward to seeing further integration with Alluxio in T3Go’s data lake ecosystem.

Building High-Performance Data Lake Using Apache Hudi and Alluxio 

Every application is built from the same type of components - buttons, inputs, dropdowns, etc., and modal windows are in this list. The reason is simple - modal windows allow us to communicate with a user and do not overwhelm a UI with additional information until it becomes needed. In this article, I will explore one of the approaches to managing modal windows in apps. Angular We use the primeng library for modals in the article's examples. Your case can be different. The general idea will be the same despite the libraries, but API can be different; be aware of it. Modal windows are just regular components - they have markup, style, and selector. But in contrast to regular components, they usually don't have to be shown on the page immediately. We need them only for the result of users' actions - confirm, discard, fill a form, etc. That's why we won't include them in our build - we will download them dynamically only if we require them. Let's see an example: public openModal(): void {
    const modalComponent = async (): Promise<any> => import('./path-to-your-modal');
    this.modalService.open(modalComponent());
} Here, we have the dynamic import approach. It means that our component will be loaded when we invoke the method. It's quite useful because we don't load the modal immediately in our chunk - only when the user needs it. This approach can help you to slightly reduce the bundle size, especially if your modals are full of logic and components. openModal In general, this is the working solution. But we can go further. Let's make a service that will handle the dynamic imports and return an instance of the modal: import { Injectable } from '@angular/core';
import { DialogService } from 'primeng/dynamicdialog';

@Injectable({ providedIn: 'root' })
export class CustomModalService {
    public modalWindowComponents: { [key: string]: () => Promise<any> } = {
        ConfirmModalComponent: () =>
            import('your-path/to-component').then((component) => component.ConfirmModalComponent),
    };

constructor(private modalService: DialogService) {}

    public async openDynamicDialogModal({
        componentName,
        data,
    }: {
        componentName: string;
        data?: { [key: string]: any };
    }): Promise<any> {
        const modalComponent = this.modalService.open(await this.modalWindowComponents[componentName](), { ...data });

        modalComponent.onClose.subscribe(() => {
            // some logic
        });
        return modalComponent;
    }
} Here, we can see several important things: We use an object to store modal windows imports where the key is the component name, and the value is an import function that returns Promise; Our method is taking the params object with the component's name (to take the import function from the object) and some additional data, which contains whatever you want for a component or its appearing configuration. Also, we have a subscription for close events. It is more convenient to handle window closing in one place; We return a Promise with the imported component. It allows us to handle the logic for each modal. Here is an example of the usage of this service: public async handleOpeningModalWindow(): Promise<void> {
    const dialogRef = await this.primengModalService.openDynamicDialogModal({
        componentName: 'DeleteStreamModalComponent',
        data: {
           // your params
        },
    });

    dialogRef.onClose
        .pipe(
            // your pipes
        )
        .subscribe(() => {
            // your logic
        });
} Conclusion By adopting the approach outlined above, Angular developers can achieve a leaner, more manageable modal window system. This strategy not only reduces bundle size but also offers a unified method for modal control and customization. Embracing dynamic imports and service-based architecture empowers developers to optimize user experiences while maintaining code scalability and efficiency.

How to Manage Modal Windows in Angular Apps

How to Build and Deploy a Real-time Traffic Monitoring App Using Quix's Computer Vision Template

কুইক্সের কম্পিউটার ভিশন টেমপ্লেট ব্যবহার করে কীভাবে একটি রিয়েল-টাইম ট্র্যাফিক মনিটরিং অ্যাপ তৈরি এবং স্থাপন করবেন

So erstellen und implementieren Sie eine Echtzeit-Verkehrsüberwachungs-App mithilfe der Computer Vision-Vorlage von Quix

Cómo crear e implementar una aplicación de monitoreo de tráfico en tiempo real utilizando la plantilla de visión por computadora de Quix

Comment créer et déployer une application de surveillance du trafic en temps réel à l'aide du modèle de vision par ordinateur de Quix

क्विक्स के कंप्यूटर विज़न टेम्पलेट का उपयोग करके रीयल-टाइम ट्रैफ़िक मॉनिटरिंग ऐप कैसे बनाएं और तैनात करें

Quix の Computer Vision テンプレートを使用してリアルタイム トラフィック監視アプリを構築および展開する方法

Quix의 컴퓨터 비전 템플릿을 사용하여 실시간 교통 모니터링 앱을 구축하고 배포하는 방법

Como construir e implantar um aplicativo de monitoramento de tráfego em tempo real usando o modelo de visão computacional do Quix

Как создать и развернуть приложение для мониторинга трафика в реальном времени с использованием шаблона компьютерного зрения Quix

Quix'in Görüntü İşleme Şablonunu Kullanarak Gerçek Zamanlı Trafik İzleme Uygulaması Nasıl Oluşturulur ve Dağıtılır

Cách xây dựng và triển khai ứng dụng giám sát lưu lượng truy cập thời gian thực bằng mẫu thị giác máy tính của Quix

如何使用 Quix 的计算机视觉模板构建和部署实时流量监控应用程序

Flink-Doris-Connector 1.4.0 allows users to ingest a whole database ( or ) that contains thousands of tables into , a real-time analytic database, . MySQL Oracle Apache Doris in one step With built-in Flink CDC, the Connector can directly synchronize the table schema and data from the upstream source to Apache Doris, which means users no long have to write a DataStream program or pre-create mapping tables in Doris. When a Flink job starts, the Connector automatically checks for data equivalence between the source database and Apache Doris. If the data source contains tables that do not exist in Doris, the Connector will automatically create those same tables in Doris, and utilize the side outputs of Flink to facilitate the ingestion of multiple tables at once; if there is a schema change in the source, it will automatically obtain the DDL statement and make the same schema change in Doris. Quick Start : For MySQL Download the JAR file: https://github.com/apache/doris-flink-connector/releases/tag/1.4.0 Maven: <dependency>
  <groupId>org.apache.doris</groupId>
  <artifactId>flink-doris-connector-1.15</artifactId>
  
  
  <version>1.4.0</version>
</dependency> : For Oracle Download JAR file: , , Flink 1.15 Flink 1.16 Flink 1.17 How to Use It For example, to ingest a whole MySQL into Doris (the MySQL table names start with or ), simply execute the following command (no need to create the tables in Doris in advance): database mysql_db tbl test <FLINK_HOME>/bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.16-1.4.0.jar \
    mysql-sync-database \
    --database test_db \
    --mysql-conf hostname=127.0.0.1 \
    --mysql-conf username=root \
    --mysql-conf password=123456 \
    --mysql-conf database-name=mysql_db \
    --including-tables "tbl|test.*" \
    --sink-conf fenodes=127.0.0.1:8030 \
    --sink-conf username=root \
    --sink-conf password=123456 \
    --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
    --sink-conf sink.label-prefix=label1 \
    --table-conf replication_num=1 To ingest an Oracle database: please refer to the . sample code How It Performs When it comes to synchronizing a whole database (containing hundreds or even thousands of tables, active or inactive), most users want it to be done within seconds. So we tested the Connector to see if it came up to scratch: 1000 MySQL tables, each having 100 fields. All tables were active (which meant they were continuously updated and each data writing involved over a hundred rows) Flink job checkpoint: 10s Under pressure test, the system showed high stability, with key metrics as follows: According to feedback from early adopters, the Connector has also delivered high performance and system stability in 10,000-table database synchronization in their production environment. This proves that the combination of Apache Doris and Flink CDC is capable of large-scale data synchronization with high efficiency and reliability. How It Benefits Data Engineers Engineers no longer have to worry about table creation or table schema maintenance, saving them days of tedious and error-prone work. Previously in Flink CDC, you need to create a Flink job for each table and build a log parsing link at the source end, but now with whole-database ingestion, resource consumption in the source database is largely reduced. It is also a unified solution for incremental updates and full updates. Other Features Joining dimension table and fact table The common practice is to put dimension tables in Doris and run join queries via the real-time stream of Flink. Based on the Async I/O of Flink, Flink-Doris-Connector 1.4.0 implements asynchronous Lookup Join, so the Flink real-time stream won't be blocked due to queries. Also, the Connector allows you to combine multiple queries into one big query, and send it to Doris at once for processing. This improves the efficiency and throughput of such join queries. Thrift SDK We introduced Thrift-Service SDK into the Connector so users no longer have to use Thrift plug-ins or configure a Thrift environment in the compilation. This makes the compilation process much simpler. On-Demand Stream Load During data synchronization, when there is no new data ingestion, no Stream Load requests will be issued. This avoids unnecessary consumption of cluster resources. Polling of Backend Nodes For data ingestion, Doris calls a frontend node to obtain a list of the backend nodes and randomly chooses one to launch an ingestion request. That backend node will be the Coordinator. Flink-Doris-Connector 1.4.0 allows users to enable the polling mechanism, which is to have a different backend node to be the Coordinator at each Flink checkpoint to avoid too much pressure on a single backend node for a long time. Support for More Data Types In addition to the common data types, Flink-Doris-Connector 1.4.0 supports DecimalV3/DateV2/DateTimev2/Array/JSON in Doris. Example Usage Read from Apache Doris: You can read data from Doris via DataStream or FlinkSQL (bounded stream). Predicate pushdown is supported. CREATE TABLE flink_doris_source (
    name STRING,
    age INT,
    score DECIMAL(5,2)
    ) 
    WITH (
      'connector' = 'doris',
      'fenodes' = '127.0.0.1:8030',
      'table.identifier' = 'database.table',
      'username' = 'root',
      'password' = 'password',
      'doris.filter.query' = 'age=18'
);

SELECT * FROM flink_doris_source; : Join dimension table and fact table CREATE TABLE fact_table (
  `id` BIGINT,
  `name` STRING,
  `city` STRING,
  `process_time` as proctime()
) WITH (
  'connector' = 'kafka',
  ...
);

create table dim_city(
  `city` STRING,
  `level` INT ,
  `province` STRING,
  `country` STRING
) WITH (
  'connector' = 'doris',
  'fenodes' = '127.0.0.1:8030',
  'jdbc-url' = 'jdbc:mysql://127.0.0.1:9030',
  'lookup.jdbc.async' = 'true',
  'table.identifier' = 'dim.dim_city',
  'username' = 'root',
  'password' = ''
);

SELECT a.id, a.name, a.city, c.province, c.country,c.level 
FROM fact_table a
LEFT JOIN dim_city FOR SYSTEM_TIME AS OF a.process_time AS c
ON a.city = c.city : Write to Apache Doris CREATE TABLE doris_sink (
    name STRING,
    age INT,
    score DECIMAL(5,2)
    ) 
    WITH (
      'connector' = 'doris',
      'fenodes' = '127.0.0.1:8030',
      'table.identifier' = 'database.table',
      'username' = 'root',
      'password' = '',
      'sink.label-prefix' = 'doris_label',
      //json write-in
      'sink.properties.format' = 'json',
      'sink.properties.read_json_by_line' = 'true'
);

Auto-Synchronization of an Entire MySQL Database for Data Analysis

This is a part of the digital transformation of a real estate giant. For the sake of confidentiality, I'm not going to reveal any business data, but you'll get a detailed view of our data warehouse and our optimization strategies. Now let's get started. Architecture Logically, our data architecture can be divided into four parts. : This is supported by Flink CDC, DataX, and the Multi-Catalog feature of Apache Doris. Data integration : We use Apache Dolphinscheduler for script lifecycle management, privileges in multi-tenancy management, and data quality monitoring. Data management : We use Grafana, Prometheus, and Loki to monitor component resources and logs. Alerting : This is where BI tools step in for user interaction, such as data queries and analysis. Data services 1. Tables We create our dimension tables and fact tables centering each operating entity in business, including customers, houses, etc. If there are a series of activities involving the same operating entity, they should be recorded by one field. (This is a lesson learned from our previous chaotic data management system.) 2. Layers Our data warehouse is divided into five conceptual layers. We use Apache Doris and Apache DolphinScheduler to schedule the DAG scripts between these layers. Every day, the layers go through an overall update besides incremental updates in case of changes in historical status fields or incomplete data synchronization of ODS tables. 3. Incremental Update Strategies (1) Set where >= "activity time -1 day or -1 hour" instead of where >= "activity time The reason for doing so is to prevent data drift caused by the time gap of scheduling scripts. Let's say, with the execution interval set to 10 min, suppose that the script is executed at 23:58:00 and a new piece of data arrives at 23:59:00, if we set , that piece of data of the day will be missed. where >= "activity time (2) Fetch the ID of the largest primary key of the table before every script execution, store the ID in the auxiliary table, and set where >= "ID in auxiliary table" This is to avoid data duplication. Data duplication might happen if you use the Unique Key model of Apache Doris and designate a set of primary keys, because if there are any changes in the primary keys in the source table, the changes will be recorded and the relevant data will be loaded. This method can fix that, but it is only applicable when the source tables have auto-increment primary keys. (3) Partition the tables As for time-based auto-increment data such as log tables, there might be less changes in historical data and status, but the data volume is large, so there could be huge computing pressure on overall updates and snapshot creation. Hence, it is better to partition such tables, so for each incremental update, we only need to replace one partition. (You might need to watch out for data drift, too.) 4. Overall Update Strategies (1) Truncate Table Clear out the table and then ingest all data from the source table into it. This is applicable for small tables and scenarios with no user activity in wee hours. (2) ALTER TABLE tbl1 REPLACE WITH TABLE tbl2 This is an atomic operation and it is advisable for large tables. Every time before executing a script, we create a temporary table with the same schema, load all data into it, and replace the original table with it. Application : every minute ETL job : 8 nodes, 2 frontends, 8 backends, hybrid deployment Configuration for first-time deployment : 32C * 60GB * 2TB SSD Node configuration This is our configuration for TBs of legacy data and GBs of incremental data. You can use it as a reference and scale your cluster on this basis. Deployment of Apache Doris is simple. You don't need other components. To integrate offline data and log data, we use DataX, which supports CSV format and readers of many relational databases, and Apache Doris provides a DataX-Doris-Writer. We use Flink CDC to synchronize data from source tables. Then we aggregate the real-time metrics utilizing the Materialized View or the Aggregate Model of Apache Doris. Since we only have to process part of the metrics in a real-time manner and we don't want to generate too many database connections, we use one Flink job to maintain multiple CDC source tables. This is realized by the multi-source merging and full database sync features of Dinky, or you can implement a Flink DataStream multi-source merging task yourself. It is noteworthy that Flink CDC and Apache Doris support Schema Change. EXECUTE CDCSOURCE demo_doris WITH (
  'connector' = 'mysql-cdc',
  'hostname' = '127.0.0.1',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'checkpoint' = '10000',
  'scan.startup.mode' = 'initial',
  'parallelism' = '1',
  'table-name' = 'ods.ods_*,ods.ods_*',
  'sink.connector' = 'doris',
  'sink.fenodes' = '127.0.0.1:8030',
  'sink.username' = 'root',
  'sink.password' = '123456',
  'sink.doris.batch.size' = '1000',
  'sink.sink.max-retries' = '1',
  'sink.sink.batch.interval' = '60000',
  'sink.sink.db' = 'test',
  'sink.sink.properties.format' ='json',
  'sink.sink.properties.read_json_by_line' ='true',
  'sink.table.identifier' = '${schemaName}.${tableName}',
  'sink.sink.label-prefix' = '${schemaName}_${tableName}_1'
); We use SQL scripts or "Shell + SQL" scripts, and we perform script lifecycle management. At the ODS layer, we write a general DataX job file and pass parameters for each source table ingestion, instead of writing a DataX job for each source table. In this way, we make things much easier to maintain. We manage the ETL scripts of Apache Doris on DolphinScheduler, where we also conduct version control. In case of any errors in the production environment, we can always rollback. After ingesting data with ETL scripts, we create a page in our reporting tool. We assign different privileges to different accounts using SQL, including the privilege of modifying rows, fields, and global dictionary. Apache Doris supports privilege control over accounts, which works the same as that in MySQL. We also use Apache Doris data backup for disaster recovery, Apache Doris audit logs to monitor SQL execution efficiency, Grafana+Loki for cluster metric alerts, and Supervisor to monitor the daemon processes of node components. Optimization 1. Data Ingestion We use DataX to Stream Load offline data. It allows us to adjust the size of each batch. The Stream Load method returns results synchronously, which meets the needs of our architecture. If we execute asynchronous data import using DolphinScheduler, the system might assume that the script has been executed, and that can cause a messup. If you use a different method, we recommend that you execute in the shell script, and check the regex filtering status to see if the ingestion succeeds. show load 2. Data Model We adopt the Unique Key model of Apache Doris for most of our tables. The Unique Key model ensures idempotence of data scripts and effectively avoids upstream data duplication. 3. Reading External Data We use the Multi-Catalog feature of Apache Doris to connect to external data sources. It allows us to create mappings of external data at the Catalog level. 4. Query Optimization We suggest that you put the most frequently used fields of non-character types (such as int and where clauses) in the first 36 bytes, so you can filter these fields within milliseconds in point queries. 5. Data Dictionary For us, it is important to create a data dictionary because it largely reduces personnel communication costs, which can be a headache when you have a big team. We use the in Apache Doris to generate a data dictionary. With it, we can quickly grasp the whole picture of the tables and fields and thus increase development efficiency. information_schema Performance : Within minutes Offline data ingestion time : For tables containing over 100 million rows, Apache Doris responds to ad-hoc queries within one second, and complicated queries in five seconds. Query latency : It only takes up a small number of servers to build this data warehouse. The 70% compression ratio of Apache Doris saves us lots of storage resources. Resource consumption Experience and Conclusion Actually, before we evolved into our current data architecture, we tried Hive, Spark and Hadoop to build an offline data warehouse. It turned out that Hadoop was overkill for a traditional company like us since we didn't have too much data to process. It is important to find the component that suits you most. On the other hand, to smoothen our  big data transition, we need to make our data platform as simple as possible in terms of usage and maintenance. That's why we landed on Apache Doris. It is compatible with MySQL protocol and provides a rich collection of functions so we don't have to develop our own UDFs. Also, it is composed of only two types of processes: frontends and backends, so it is easy to scale and track.

Building a Data Warehouse for Traditional Industry: Hadoop is Overkill

In Vue, data is typically passed from parent components to their children in a uni-directional fashion. This is transferred with , which are the properties or attributes we give to . props components For example, if we call a component , which has a prop called , that property would become available within the component itself, letting us do what we want to do with it. That way, the data is passed down to the child component when we declare it in the parent component or page: PageOne name name PageOne In most scenarios, props allow us to do everything we need to do with data. - from a child component to its parent. For this, we use , which lets us send data upwards, and then trigger an event in the parent component should an event be fired. Sometimes, however, we need to emit data upwards $emit $emit How works in Vue? $emit There are three ways to fire in Vue, depending on if you're using the Options API, Composition API, or inlining your events. $emit $emit within the Options API. this.$emit if used in your HTML template. $emit and if used in the Composition API. defineEmits emit . Let's say we have a counter component, which looks like this: Let's take a look at how this works, through a silly example <template>
    <button @click="$emit('counterEvent')">Click Me</button>
</template> This component is stored in a file called . Our component can't be changed as it's used in other places, but it does have an event fired any time it is clicked. This is perfect since we can use this in our parent component. Counter.vue $emit So what if we want to add this component somewhere - for example, in our file - and use it to display the value of our counter. Let's try doing that now: App.vue <template>
    <h1>{{ counter }}</h1>
  	<Counter @counter-event="incrCounter"/>
</template>

<script>
import Counter from './Counter.vue'
export default {
    // Add our components
  	components: {
      Counter
    },
    // Store our data
    data() {
        return {
            counter: 0
        }
    },
    methods: {
        incrCounter: function() {
            this.counter += 1;
        }
    }
}
</script> Let's break this down - first of all, we include our . Since there is an event called , we can attach that to our HTML. Whenever fires, it'll fire the , and thus the function within that property. anytime fires. Counter $emit counterEvent Counter $emit counterEvent Here, we run incrCounter counterEvent By doing that, we can also increase our data by 1, since that is what does. As such, we've emitted the click event upwards to our parent component. counter incrCounter Why did we use Kebab Case? You may notice that when we defined our event, we used camel case ( ), but when tracking the event, we used kebab case ( ). $emit counterEvent counter-event In , it is fine to use and interchangably since Vue 3 automatically converts to . In , this functionality does not exist, so just stick with for both. Vue 3 counterEvent counter-event counterEvent counter-event Vue 2 counter-event Passing Data with $emit Let's say instead, we want our components to define how much the should increase. If we want to do that, we can pass a second argument to the function, which is the value: counterEvent $emit <template>
    <button @click="$emit('counterEvent', 2)">Click Me</button>
</template> , we are passing the value to our . Let's go back to our file. To leverage this value in , we need to write it as a function. Below, is the value: Here 2 counterEvent App.vue counterEvent n <template>
    <h1>{{ counter }}</h1>
  	<Counter @counter-event="(n) => incrCounter(n)"/>
</template>

<script>
import Counter from './Counter.vue'
export default {
    // Add our components
  	components: {
      Counter
    },
    // Store our data
    data() {
        return {
            counter: 0
        }
    },
    methods: {
        incrCounter: function(value) {
            this.counter += value;
        }
    }
}
</script> Now, our counter will increase by the value put in the child component, allowing us to pass data to our parent component as well. As you'd expect, this is not limited to just numbers but can include any data structure - including objects and strings. Using with the Options API $emit We've shown a quite simple example, but we could also have written our child component using a function instead. Here is an example with the , using : Counter.vue Options API this.$emit <template>
    <button @click="emitFunction">Click Me</button>
</template>

<script>
export default {
  emits: [ 'counterEvent' ],
 	methods: {
		emitFunction: function() {
		    this.$emit('counterEvent', 2)
        }
    }
}
</script> This may prove to be a slightly cleaner way to use , especially if you want to do other things along with using whenever a button is clicked. $emit $emit Adding your emit events to your prototype You may note that we also defined our emit event in on the prototype. This is good practice for two reasons: emits by showing which emit events are possible in this component. It lets you self-document the code , since Vue will throw an error if an emit event is used but not found in the array. It helps you keep track of deprecated emits emits Using with the Composition API $emit We can use with the Composition API - the only difference is we have to use instead. $emit defineEmits <template>
    <button @click="emitFunction">Click Me</button>
</template>

<script setup>
import { defineEmits } from 'vue'

const emit = defineEmits(['counterEvent']);
const emitFunction = function() {
    emit('counterEvent', 2)
}
</script> is used to define a full list of all permissible events. Here, we only have one, . If you had more than one, you could define them as so: defineEmits emit counterEvent const emit = defineEmits(['counterEvent', 'anotherEvent', 'finalEvent']); If you use an emit event not listed in , Vue will throw a warning, similar to using on the Options API. Otherwise, you can then use the function to emit as usual, without needing to use the at all. defineEmits emits emit() Options API Final Thoughts and Best Practices Emit is a powerful tool for sending data back up to the parent when we have to. This means datastreams can be two-way in Vue. When defining code, the two main best practices are: emit Always define your emit events in either or , which will help you keep your code clean and well documented. emits defineEmits Normal convention in Vue 3 is to use kebab case ( ) for HTML, and camel case ( ) in script. As such, it is best to follow this convention here too. this-is-kebab-case thisIsCamelCase I hope you've enjoyed this guide on how works. Stay tuned for more . $emit Vue Content Also Published Here

Learn How to Emit Custom Events in Vue With $emit

Over the past decade, 3D sensors have emerged to become one of the most versatile and ubiquitous types of sensor used in robotics. In many robotic applications, 3D sensing has become the de facto choice for tasks such as near-field object detection and collision avoidance, surface and object inspection, and map creation. There are a number of 3D sensing modalities available. In this roundup, and we’ll be covering the three most common that we work with at Tangram Vision: Stereo CMOS (both active and passive), structured light, and time of flight. We won’t be covering LiDAR, despite the fact that LiDAR data is also three dimensional. LiDAR users, fret not…we’ll be covering LiDAR as its own separate class of sensors in a separate post in the future! 1. Stereo 3D Sensors These ubiquitous sensors bring a combination of good overall performance coupled with low cost. Stereo 3D sensors come in two basic types: passive and active. Structure Core Active Stereo 3D Sensor. Image attribution: , Inc. Occipital Passive stereo 3D sensors are the least expensive of all 3D sensors, as they use off-the-shelf components that are inexpensive to source and manufacture. In addition, they are available in a wide range of baselines, giving users a choice of sensing ranges that are best adapted to their use case. However, most rely on visible light to operate effectively, which means they struggle to perform in low-light or unlit conditions. Active stereo 3D sensors add an infrared (IR) pattern projector which both increases the fidelity of 3D data captured, as well as adding more reliable operation in lower light conditions. However, the IR projectors on these sensors are limited in range, which relegates these sensors to near- and mid-range applications, regardless of baseline. The addition of an IR pattern projector effectively turns these sensors into hybrid stereo/structured light sensors (more on structured light 3D sensors in the next paragraph!). Examples of stereo 3D sensors: — Passive stereo 3D sensors available in large and small form factors, with wide and narrow baselines. The Zed Mini and Zed 2 feature integrated IMUs for kinetic data streams. Stereolabs Zed - Industrial-grade stereo 3D sensors with ruggedized construction and Gigabit Ethernet connections. Ensenso — Active stereo 3D sensor with bespoke IR projector and built-in IMU for kinetic data. Available as an embeddable module or as an encased standalone sensor. Occipital Structure Core — The D series (D415, D435, D435i, D455) are a diverse range of low cost active stereo 3D sensors. The D435i adds a built-in IMU for kinetic data, while the newest D455 adds extended range that is suitable for mid-range robotic navigation. Intel RealSense 2. Structured Light Sensors Structured light is the most ubiquitous 3D sensing modality used in robotics, thanks to the popularity of the original PrimeSense 3D sensors, which included the first generation Microsoft Kinect. Microsoft Kinect for Xbox 360. By Evan Amos — Own work, Public Domain, https://commons.wikimedia.org/w/index.php?curid=16059165 Structured light 3D sensors combine low cost with high-fidelity 3D data capture, as well as performance in a wide set of lighting conditions, with one notable exception: direct sunlight, or bright indirect sunlight. This is because the infrared light source used by structured light sensors gets overpowered by infrared light in the same bandwidth that is naturally present in sunlight. Like stereo 3D sensors, structured light sensors are available in an array of small to large baselines that are suited to tasks that occur at different ranges from the sensor. Examples: — A low-cost structured light 3D sensor that is often used as a drop-in substitute for the original Microsoft Kinect. Orbbec Astra (No longer available new) — The first standalone 3D sensor designed specifically to be used with mobile devices like Apple iPad and iPhone models. Compatible with for depth-powered application development. Now replaced by the active stereo sensor. Occipital Structure Sensor OpenNI Occipital Structure Sensor Mark II (No longer available new) — The most ubiquitous consumer-grade 3D sensor ever produced. Originally launched to support Xbox gaming, the first Kinect became popular with roboticists and hackers for adding robust 3D sensing at low cost to robots, drones and other platforms. Compatible with for depth-powered application development. Microsoft Kinect V1 OpenNI (No longer available new) — The Carmine series is near-identical in specification to the original Microsoft Kinect, and was available in short- and mid-range models. Unlike the original Microsoft Kinect, the Carmine does not include a microphone array. Compatible with for depth-powered application development. PrimeSense Carmine OpenNI — Ruggedized 3D sensors for industrial applications. The Photoneo phoXi line is available in multiple baselines. Photoneo — A range of structured light 3D sensors available in multiple baselines, with USB3 connection for the Zivid One series, and CAT 6A Ethernet connection for the Zivid Two. Zivid 3. Time of Flight Sensors Time of flight (ToF) sensors are effectively LiDAR sensors — aka, light radar. They send out packets of infrared light, and record the time it takes for the infrared light sent out to return. Despite using infrared light sources like structured light sensors, the manner in which ToF sensors transmit and receive light make them less susceptible to interference from bright and indirect sunlight. PMD Time of Flight (ToF) 3D sensor. ToFExpert, , via Wikimedia Commons CC BY-SA 3.0 ToF sensors often have wider range capabilities than structured light sensors, and can operate accurately for longer distance use cases. However, they aren’t able to return 3D images with the fidelity offered by structured light or stereo 3D sensors. Therefore, they are better suited for tasks like navigation assistance versus surface inspections. Examples: — The newest member of the Microsoft Kinect family of 3D sensors is a compact ToF module that adds a 12MP RGB video camera that is registered to the depth stream for high quality color imaging and textures. The Azure Kinect includes a built-in IMU for kinetic data. Microsoft Azure Kinect — Simple, compact ToF 3D sensors that can be purchased as embeddable modules, or as enclosed standalone sensors. PMD Pico — Increasingly hard-to-find. However, like the original Microsoft Kinect, PrimeSense Carmine and original Occipital Structure Sensor 3D sensors, the Xtion 2 is compatible with for depth-powered application development. ASUS Xtion 2 OpenNI Comparing Stereo Sensors to Structured Light to ToF Perception Sensors So why choose one modality over another? Well, we provided hints in the above descriptions for each modality, but let’s get more concrete on why you might choose one over another. At the highest level, the primary factors that determine 3D sensing selection are: Sensor Performance Impact on Host Environmental Limitations Let’s explore each of these factors in detail, and then show how each sensing modality compares. Sensor Performance The main parameters we’ll be exploring here are range, resolution and reliability. Range is perhaps the most important parameter, because different robot tasks are defined by the range in which they occur. For example, mapping environments at high speed for autonomous vehicles is typically left to rotating LiDAR units, as they have ranges that can exceed 100 meters, which allows them to see distant obstacles well before the autonomous vehicle is upon those obstacles. Those same LiDAR units, however, are poorly suited for close visual inspections, as these tasks benefit from sensors with sub-meter ranges that can support enhanced detail capture and accuracy. For stereo and structured light 3D sensors, range is a function of the baseline between the optical elements on the sensor itself. The rule of thumb is that the wider the baseline, the longer the range. Similarly, a narrower baseline creates a shorter range. So why not just spec a 3D sensor with the widest baseline possible? Well, for one, form factor limitations can preclude this approach. But, more importantly, wider baseline sensors’ sensing ranges start further away from the sensor. That means that a wide baseline sensor might ignore objects that are less than two meters from the sensor. A shorter baseline sensor might not have as long of a range, but it might be able to start sensing within .25 meters of the sensor. Therefore, baseline and the resultant range becomes a tradeoff of how near you need to sense versus how distant you need to sense. You get one or the other, but you can’t get both. Most 3D sensors have ranges that start at around 0.5–1m from the sensor, and extend to 3–6m at best. Keep in mind that at the furthest reaches of a sensor’s range, resolution and accuracy can start to diminish rapidly. ToF 3D sensors typically have much greater range than structured light or stereo 3D sensors, and can often receive reliable signals from objects that are up to 25m away. However, ToF sensors often suffer from excessive noise, and resolution can be much lower than similarly priced 3D modalities. Speaking of resolution, this is both a function of range, as well as the nature of the optics used by the sensor. Structured light sensors’ resolution is limited by the density of the pattern that they project (which gets less dense as distance increases), and the resolution of the IR camera that reads the pattern as it falls on objects and environments to be sensed. Stereo sensors’ resolution is limited by the native resolution of the cameras used for the sensor. For tasks such as navigation, high resolution isn’t strictly necessary, unless objects that might cause an obstruction are very fine (an extension cable, for instance, may be on the edge of detectability for some low-resolution 3D sensors). For tasks such as surface inspections and 3D modeling, high resolution is much more desirable. Because these tasks often occur at close range, many structured light and stereo sensors are well suited to this kind of task. ToF sensors are best suited for mid-distance navigation tasks, or skeletal tracking. These tasks don’t require high-resolution sensing, but benefit from the range and operational speed of ToF sensors. Lastly, let’s talk about reliability. For sensors, this comes in two forms: data acquisition reliability, and mechanical reliability. Data acquisition reliability comes down to the inherent limitations of each sensor modality. Stereo sensors, for instance, need sufficient ambient light in the visible spectrum to work properly. Conversely, infrared-based sensors like structured light and ToF suffer when exposed to too much ambient light, causing sensor washout and data integrity issues. These sensors also struggle with black surfaces (they absorb IR wavelengths) and shiny surfaces (they scatter projected IR light). Active stereo sensors with infrared pattern projectors offer a “best of both worlds” approach that broaden sensor reliability over a number of environments and lighting conditions. Mechanical reliability is often the result of compromises inherent to a specific modality, or result from manufacturer cost cutting. Many consumer-grade sensors use USB for data and power transfer; this is fine in a lab or test environment. However, in real-world use, USB can be flaky and error-prone, resulting in intermittent sensor failures and compromised data. Ethernet is much more stable, but is rarely found on consumer-grade sensors. It is much more likely to be found on much more expensive industrial-grade sensors, where it is a prerequisite for a purchase decision. Impact on Host The main parameters we’ll be exploring here are power consumption, compute use, heat dissipation, data transmission and form factor. All robotic hosts have finite amounts of compute and power available. Robots, drones and other vision-enabled devices are often resource balancing acts where multiple sensors, motors, actuators, and other components fight for scarce system resources. Different sensors have different impacts. Sensors with active IR components are much more power hungry than passive stereo 3d sensors, for instance. But passive stereo sensors with high dynamic range (HDR) cameras may use much more compute than a structured light sensor, as the images they capture are much more data rich. Thus, sensor decisions must be made holistically in context of the other components that require system resources (and that subject is complex enough to deserve its own blog post). Similarly, data transmission and storage can often be limited on robots and drones. Again, the example of using an HDR-equipped 3D sensor can strain even robust WiFi networks given the sheer size of the datastreams being captured and transmitted. Lastly, there are industrial design considerations that can limit what sensor options can be specified. These considerations include the size and weight of a sensor (particularly important for lightweight devices like drones), as well as a sensor’s ability to dissipate heat generated. The latter is of particular importance to IR-equipped sensors. The laser-based IR components in these sensors rely on operating within a specific temperature range to maintain the bandwidth at which they operate. Deviating by even a few degrees Fahrenheit outside of the operating temperature range can cause significant variations in sensor readings, corrupting the data generated enough to make it useless. Therefore, considerations must be made to ventilate or insulate these sensors, depending on the environments in which they will be used. Environmental Limitations The main parameters we’ll be exploring here are ambient lighting, and connectivity. We mentioned above that many of these sensors are sensitive to ambient lighting, but it bears repeating. Lighting challenges are one of the most common causes of 3D sensor failures for robots. A robot that worked exceptionally well in a laboratory environment with consistent lighting conditions will suddenly experience drastic and intermittent failures when placed in a real world context with changing, inconsistent lighting conditions. Therefore, it is important to know all the lighting permutations that may exist upon deployment, and account for those during sensor selection. Connectivity is also important. If sensor data is to be processed and used locally, then connectivity isn’t an issue. However, for robots that will be deployed in an environment with low bandwidth, then it is either necessary to compress or filter sensor data on the host before transmitting, or it is necessary to choose a low bandwidth sensor type (for instance, ToF). Again, it is important to test for packet size thresholds under real world bandwidth constraints, as laboratory environments may not sufficiently reveal deficiencies. Synthesizing the Data and Deciding We’ve put together a handy chart showing different , and the 3D sensors that work best for those scenarios. We’ve noted what key decision factors should be considered for each application, and added notes to further clarify why specific choices should be made. robot tasks to be done We hope that you’ve found this guide to 3D sensors useful! Our next sensor roundup will cover LiDAR sensors, so stay tuned. If you’re currently working on a vision-enabled product like a robot or a drone, please take a look at what we do at . We can help you get your product to market faster, and keep it functioning more reliably upon deployment, too. Tangram Vision Also published at https://medium.com/tangram-visions/sensors-101-3d-sensing-66e528f276c4

3 Common Types of 3D Sensors: Stereo, Structured Light, and ToF

The OWASP Top 10 is a standard awareness document for developers and web application security. It represents a broad consensus about the most critical security risks to web applications. In this article, I want to cover the second part of the TOP 10 vulnerabilities and how to protect against them using .NET. 6. Security Misconfiguration Security misconfiguration can happen at any level of an application stack, including the network services, platform, web server, application server, database, frameworks, custom code, and pre-installed virtual machines, containers, or storage. Such flaws frequently give attackers unauthorized access to some system data or functionality. What to do? (When using TLS) Redirect a request made over Http to https: app.UseHttpsRedirection(); //Startup.cs Send the anti-forgery token with every POST/PUT request: To automatically validate all requests other than GET, HEAD, OPTIONS and TRACE you need to add a global action filter with the attribute inside your Startup.cs AutoValidateAntiforgeryToken services.AddMvc(options =>
{
    options.Filters.Add( AutoValidateAntiforgeryTokenAttribute());
}); new If you need to also validate the token on GET, HEAD, OPTIONS or TRACE - requests you can add the attribute to the controller method (for MVC controllers) or parent class (for Razor pages): ValidateAntiforgeryToken [ ]
[ ] HttpGet ValidateAntiforgeryToken IActionResult ( ) public DoSomethingDangerous [ ]
[ ] : HttpGet ValidateAntiforgeryToken public class SafeModel PageModel In case you can't use a global action filter, add the attribute to your controller classes or razor page models: AutoValidateAntiforgeryToken [ ] AutoValidateAntiforgeryToken public class UserController [ ] : AutoValidateAntiforgeryToken public class SafeModel PageModel 7. Cross-Site Scripting (XSS) XSS is the second most prevalent issue in the OWASP Top 10, and is found in around two-thirds of all applications. The impact of XSS is moderate for reflected and DOM XSS, and severe for stored XSS, with remote code execution on the victim’s browser, such as stealing credentials, sessions, or delivering malware to the victim. What to do? DO NOT: Trust any data the user sends you, prefer white lists (always safe) over black lists DO NOT: Use the [AllowHTML] attribute or helper class @Html.Raw unless you really know that the content you are writing to the browser is safe and has been escaped properly. 8. Insecure Deserialization The impact of deserialization flaws cannot be overstated. These flaws can lead to remote code execution attacks, one of the most serious attacks possible. What to do? Validate User Input Malicious users are able to use objects like cookies to insert malicious information to change user roles. In some cases, hackers are able to elevate their privileges to administrator rights by using a pre-existing or cached password hash from a previous session. Prevent Deserialization of Domain Objects Run the Deserialization Code with Limited Access Permissions If a deserialized hostile object tries to initiate a system processes or access a resource within the server or the host's OS, it will be denied access and a permission flag will be raised so that a system administrator is made aware of any anomalous activity on the server. Don't allow the datastream to define the type of object that the stream will be deserialized to. You can prevent this by for example using the DataContractSerializer or XmlSerializer if at all possible. Where JSON.Net is being used make sure the TypeNameHandling is only set to None. TypeNameHandling = TypeNameHandling.None If JavaScriptSerializer is to be used then do not use it with a JavaScriptTypeResolver. If you must deserialise data streams that define their own type, then restrict the types that are allowed to be deserialized. One should be aware that this is still risky as many native .Net types potentially dangerous in themselves. 9. Using Components with Known Vulnerabilities Prevalence of this issue is very widespread. Component-heavy development patterns can lead to development teams not even understanding which components they use in their application or API, much less keeping them up to date. While some known vulnerabilities lead to only minor impacts, some of the largest breaches to date have relied on exploiting known vulnerabilities in components. What to do? Keep the .Net framework updated with the latest patches Keep your packages up to date, many will contain their own vulnerabilities. NuGet Run the against your application as part of your build process and act on any high level vulnerabilities. OWASP Dependency Checker 10. Insufficient Logging & Monitoring Attackers rely on the lack of monitoring and timely response to achieve their goals without being detected. Most successful attacks start with vulnerability probing. Allowing such probes to continue can raise the likelihood of successful exploit to nearly 100%. What to do? Ensure all login, access control failures and server-side input validation failures can be logged with sufficient user context to identify suspicious or malicious accounts. Establish effective monitoring and alerting so suspicious activities are detected and responded to in a timely fashion. Log all errors from the Startup.cs, so that anytime an error is thrown it will be logged. { (env.IsDevelopment())
    {
        _isDevelopment = ;
        app.UseDeveloperExceptionPage();
    } app.UseExceptionHandler(errorApp =>
    {
        errorApp.Run( context =>
        { errorFeature = context.Features.Get<IExceptionHandlerFeature>(); exception = errorFeature.Error;

            Log.Error(String.Format( , exception.StackTrace.ToString()));
        });
    });

    app.UseAuthentication();
    app.UseMvc();
} ( ) public void Configure IApplicationBuilder app, IHostingEnvironment env if true //Log all errors in the application async var var "Stacktrace of error: {0}" Monitor key performance indicators. In .NET a great option to add monitoring capabilities is . Application Insights Thanks for reading! - Part one https://hackernoon.com/owasp-top-10-net-protection-a-guide-part-1-w92r3wis P - Read more about OWAS https://owasp.org/www-project-top-ten/

OWASP Top 10 .NET Protection Guide (Part 2)

is one of the most versatile data streaming open-source solution that exists. It supports all the primary functions of a typical batch processing system such as SQL, Connectors to Hive, Group By, etc. while providing fault-tolerance and exactly-once semantics. Hence, you can create a multitude of push-based applications using it. Apache Flink However, one of the significant drawbacks of the Apache Flink has been the inability to modify the checkpointed state of the program. Let’s first see what I mean by that. Checkpointing Flink provides fault-tolerance by using a mechanism called checkpointing. It periodically takes a snapshot of all the stateful operators/functions of your programs and stores them in a highly durable store such as HDFS. Checkpointing allows the Flink program to resume from this snapshot. This is helpful in the cases of failures due to some error such as a simple exception not handled or a loss of data node in your YARN/Mesos/k8s cluster. This snapshot is stored in a binary format only understood by Flink, which makes it difficult to modify the state before restart. Why would you need to modify the data? There can be multiple cases where you might need only partial data from the checkpoint and may want to update the other. An example job being Reading numerical data from one Kafka topicAggregate over a window of 1 hourClassify using some statistical thresholds provided by config stored in your operator’s state. { { StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        Properties kafkaConsumerProperties = Properties();
        kafkaConsumerProperties.setProperty( , );
        kafkaConsumerProperties.setProperty( , );

        ObjectMapper objectMapper = ObjectMapper();
        FlinkKafkaConsumer010<String> kafkaConsumer010 = FlinkKafkaConsumer010<>( , SimpleStringSchema(), kafkaConsumerProperties);
        DataStream<String> kafkaSource =  env.addSource(kafkaConsumer010).name( ).uid( );
        DataStream<TestData> aggregatedStream = kafkaSource
                .map(row -> objectMapper.readValue(row, TestData.class))
                .keyBy(TestData::getKey)
                .timeWindow(Time.hours( ))
                .reduce((rowA, rowB) -> {
                   TestData result = TestData();
                    result.setKey(rowA.getKey());
                    result.setValue(rowA.getValue() + rowB.getValue());
                    result.setCreatedAt(System.currentTimeMillis()); result;
                }).name( ).uid( );

        DataStream<LabeledTestData> labeledTestDataDataStream =  aggregatedStream.keyBy(TestData::getKey).flatMap( ClassifyData()).name( ).uid( );
        labeledTestDataDataStream.map(row -> objectMapper.writeValueAsString(row)).print();
        env.execute();
    }
} public class TestCheckpointJob Exception public static void main (String[] args) throws final new "bootstrap.servers" "localhost:9092" "group.id" "test_group_id" new new "test_topic" new "kafka_source" "kafka_source" 1 new return "aggregate_stream" "aggregate_stream" new "classify_data" "classify_data" {

        ValueState<Integer> threshold; { .open(parameters);
            threshold = getRuntimeContext().getState( ValueStateDescriptor<Integer>( , Integer.class));
        } {
            LabeledTestData labeledTestData = LabeledTestData();
            labeledTestData.setKey(testData.getKey());
            labeledTestData.setValue(testData.getValue());
            labeledTestData.setCreatedAt(testData.getCreatedAt());
            String label = ; (threshold.value() != ){
                label = (testData.getValue() > threshold.value()) ? : ;
            }
            labeledTestData.setLabel(label);
            collector.collect(labeledTestData);
        }
} < , > class ClassifyData extends RichFlatMapFunction TestData LabeledTestData @Override Exception public void open (Configuration parameters) throws super new "thresholdState" @Override Exception public void flatMap (TestData testData, Collector<LabeledTestData> collector) throws new "UNCLASSIFIED" if null "L1" "L2" Suppose, your job gets killed, and now you want to restart it using a checkpoint, but you need to modify the config as well. Earlier, there was no way to do so other than to wait for the job to start with old config and than overwrite it using a stream from Kafka or filesystem. Now, however, you can easily do that using the new API. Let’s modify the above example to do so. Bootstrapping the state Following are the necessary steps required to bootstrap your state Add Dependency org.apache.flink flink-state-processor-api_2.11 1.9.0 < > dependency < > groupId </ > groupId < > artifactId </ > artifactId < > version </ > version </ > dependency This is not included in the default Flink dependency and needs to be added separately in pom.xml file. Create a Bootstrap function {
        ValueState<Integer> threshold; {
            threshold = getRuntimeContext().getState( ValueStateDescriptor<Integer>( , Integer.class));
        } {
            threshold.update(testConfig.getThresholdValue());
        }
} < , > class ConfigBootstrapper extends KeyedStateBootstrapFunction String TestConfig @Override Exception public void open (Configuration parameters) throws new "thresholdState" @Override Exception public void processElement (TestConfig testConfig, Context context) throws This function tells Flink what state to update when it receives the data. In this example, we are updating the threshold state using the TestConfig data we collect. Flow the config data {
     TestConfig testConfig = TestConfig();
        testConfig.setKey( );
        testConfig.setThresholdValue( );

        DataSet<TestConfig> configDataSet = executionEnvironment.fromElements(testConfig);
        BootstrapTransformation<TestConfig> transformation = OperatorTransformation
                .bootstrapWith(configDataSet)
                .keyBy(TestConfig::getKey)
                .transform( ConfigBootstrapper()); transformation;
} BootstrapTransformation<TestConfig> getConfigTransformation (ExecutionEnvironment executionEnvironment) new "global" 10 new return Now, you need to flow the config data. The Flink state processor API works seamlessly with Dataset API. It doesn’t imply you can’t use bootstrapping in a Stream environment. It’s just that the data for bootstrapping can only be loaded using Batch API. You can create both Batch and Stream environment in a single job. Here, I have just created a single config object and then created a Dataset on top of it. We then create a transformation. It specifies what dataset to use with Bootstrap Function. Update the save point { {
      bootstrapConfig(); }
} {
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        ExistingSavepoint existingSavepoint = Savepoint.load(executionEnvironment, , MemoryStateBackend());
        BootstrapTransformation<TestConfig> configTransformation = getConfigTransformation(executionEnvironment);
        String newSavepointPath = ;
        existingSavepoint.withOperator( , configTransformation).write(newSavepointPath);
} public class TestCheckpointJob Exception public static void main (String[] args) throws //Rest same as previous code IOException static void bootstrapConfig () throws "oldSavepointPath" new "newSavepointPath" "classify_data" Next, we load the savepoint from the old directory and then update the states of the operators. To update the state, we need to specify the UID of the operator in the streaming job and the transformation created in the step. Once that is done, we can rewrite this modified savepoint in a new path. Do note that the new path contains the shallow copies of pointers from the old path. It means, deleting the old savepoint path will corrupt the new one, and hence you should refrain from doing so. Now, you can resume your Flink job using this new savepoint path. bin/flink run -s newSavepointPath test- .jar checkpoint You can even create a new Savepoint instead of updating the old one. For that, you need to do instead of Savepoint.create() Savepoint.load() Flink’s State Processor API was one of the most requested features and now it’s finally here. The API is available only in 1.9.0 and above versions. You can explore the whole API in . the official documentation Connect with me on LinkedIn or Twitter or drop a mail to kharekartik@gmail.com

The API to Bootstrap Your Flink Jobs Has Arrived

Between its SQL and Runtime layers, Alibaba has helped optimize Apache Flink for large-scale production environments like its own This article is part of Alibaba’s Flink series . As an open-source framework for big data computing, Apache Flink has undergone extensive optimization to meet a range of users’ demands for enhancement. For Alibaba Group, where the framework is deployed in a large-scale production environment, the need for these changes has motivated its real-time computing team to contribute many of Flink’s most valuable optimizations, benefiting the Flink community and Alibaba alike. In this article, we look at Alibaba’s contributions in two key aspects of the framework’s architecture, tracing developments including Flink’s Query Processor, BinaryRow, and AsyncOperator components. Contributing Flink’s SQL Layer To enable users to develop code sets based on their own business logic that can be run by various different scenarios simultaneously, Flink first needed to be able to provide those users with a unified API. Alibaba Group’s real-time computing team determined that SQL made the best choice toward this end as a recognized classic that has benefited from decades of testing in the field of batch processing. In stream computing, the concept of dynamic tables has recently emerged as a way of describing stream computing that uses SQL as batch processing would, based on equivalent logic. In this way, users can use SQL to describe their own business logic, and a given query statement can either be a batch task or a high-throughput low-latency stream computing task when executed. Moreover, processing for such statements can be carried out such that historical data is first computed using batch processing technology before its processing is automatically converted to a stream computing task to deal with the latest real-time data. Under the SQL declarative API, the engine has a greater number of choices and greater room for optimization. The following sections look at several of the most important such optimizations in detail. Optimizing Flink’s SQL architecture with Query Processor Alibaba’s first key optimizations to Flink upgraded and replaced its SQL layer architecture. Developers familiar with Flink through research or actual use will likely be aware that Flink has two basic APIs: one for DataStream (provided for streaming users), and one for DataSet (provided for batch users). These two sets of APIs have completely different execution paths, as well as needing to generate different tasks to execute. After a given series of optimizations, Flink’s native SQL layer will call the DataSet API or the DataStream API, depending on the user’s requirements for stream or batch processing. This often results in the user facing two sets of predominantly independent technology stacks in daily development and optimization, meaning that many things need to be repeated twice. This will also result in optimization on one side of the technology stack that is not available on the other side. To counter these issues, Alibaba proposed a new Query Processor at the SQL layer, consisting primarily of an optimization layer (QueryOptimizer) that can be reused by streams and batches as much as possible and a Query Executor based on the same StreamOperator interface. Using this model, more than 80 percent of work including common optimization rules and underlying data structures can be reused on both technology stacks. Simultaneously, streams and batches still retain their own specific optimizations and operators to satisfy different jobs. Optimizing Flink’s data structure with BinaryRow Having unified the SQL layer’s technical architecture, Alibaba began to seek a more efficient underlying data structure that would make Blink more efficient in the SQL layer. In the native Flink SQL, a data structure called Row is used uniformly. It consists of some Java objects and represents a row in a relational database. For example, if the current row of data consists of an integer, a float, and a string, then Row will contain a Java integer, a Java double and a Java string. As developers know well, these Java objects have a lot of overhead in heap, and also introduce unnecessary boxing and unboxing operations when accessing the data. To counter these problems, Alibaba proposed a new data structure called BinaryRow, which like the original Row represents a row in a relational database but instead stores the data in binary. Continuing with the previous example, these three different type fields are uniformly represented by Java’s byte[], bringing numerous benefits. First, a great deal of unnecessary consumption is removed from the storage space, making storage of the object more compact. Second, many unnecessary serialization and deserialization overheads can be avoided when dealing with network or state storage. Finally, the entire execution code is made friendlier to the GC after removal of all unnecessary boxing and unboxing operations. With this new, efficient underlying data structure, the SQL layer’s overall execution efficiency more than doubles. Optimizing implementation level code generation At the operator’s implementation level, Alibaba has introduced a wider ranger of code generation techniques to Flink. The unification of Flink’s technical architecture and underlying data structure achieves a wider range of reuse for many code generation techniques. Thanks to SQL’s strong type guarantee, Flink can meanwhile recognize the type of data the operator will need to process in advance and thus generate more specific and efficient execution codes. In the native Flink SQL, only simple expressions like a>2 or c+d can apply code generation techniques. Due to Alibaba’s optimization, some operators are able to perform overall code generation such as for sorting and aggregation, which allows the user to more flexibly control the operator’s logic or to directly embed the final running code into the class, and thus eliminate expensive overhead for function calls. Further, some underlying data structures and algorithms that apply code generation techniques like sorting algorithms, binary data-based HashMap, and so on can also be shared and reused between the operators of streams and batches, allowing the user to truly enjoy the benefits of unification between technology and architecture. By optimizing the data structures and algorithms for certain batch processing scenarios, it also becomes possible to improve stream computing performance. The optimizations discussed so far have all focused on changes to Flink’s SQL layer. The following sections look at important improvements Alibaba has contributed to Flink’s Runtime layer. Improving Flink’s Runtime Layer Alibaba’s real-time computing team predictably discovered a number of challenges in enabling Flink to take root in Alibaba’s large-scale production environment. Enabling large-scale cluster deployment Ahead of other issues, integrating Flink with other cluster management systems presented a series of consecutive problems, due to the fact that Flink’s native cluster management model was not yet complete and could not use other more mature cluster management systems natively. Specifically, Alibaba needed to answer questions of how to coordinate resources among multi-tenants, how to dynamically apply for and release resources, and how to specify different resource types. To solve these challenges, Alibaba’s real-time computing team performed extensive research and analysis before ultimately choosing to transform Flink’s resource dispatch system so that Flink can run on the Yarn cluster natively. Further, the team refactored the Master architecture to make a given Job correspond to a specific Master, such that the Master no longer generates a cluster bottleneck. Based on this vantage, Alibaba and the Flink community then jointly launched a new Flip-6 architecture that turns Flink resource management into a pluggable architecture, thus laying a sound foundation for Flink’s sustainable development. As a result, Flink can now seamlessly run on YARN, Mesos, and K8s, offering a powerful demonstration of the value of this architecture. Improving Flink’s FailOver mechanism After solving the above challenges, Alibaba looked next to improve Flink’s reliability and stability, focusing on its FailOver mechanism to ensure high availability in the production environment. The first step in doing so was to improve Flink’s Master FailOver. Flink’s native Master FailOver restarts all jobs, while Alibaba’s improvements ensured that Master FailOver would not impact normal job running in any way. Next, Alibaba introduced a Region-based Task FailOver to minimize user impact in the event of Task FailOver. With these improvements, a large number of Alibaba’s service clients have begun to migrate real-time computing to Flink. Removing the Checkpoint bottleneck Flink’s capacity for stateful streaming is one of its greatest advantages, as its Chandy-Lamport algorithm-based Checkpoint mechanism enables its exactly-once consistency and computing power. However, early Flink versions suffered in Checkpoint performance from a particular bottleneck under large-scale volumes of data, which Alibaba sought to resolve through optimization to Checkpoint. First, Alibaba developed an incremental Checkpoint mechanism to deal with the dozens of terabytes of State data generated by big jobs in Alibaba’s production environment, given that implementing a full checkpoint would lead to surging costs. With the incremental Checkpoint mechanism, the cost of checkpoint is reduced to a constant but small flow. Secondly, Alibaba developed a model for merging small files in Checkpoint, due to problems caused by a rising number of checkpoint files. As the number of Flink jobs grows in the entire cluster, the number of checkpoint files eventually overwhelms the HDFS NameNode. Alibaba’s approach reduces pressure on the NameNode by a factor of several dozen through organized merging of numerous smaller checkpoint files into a single large file. Enabling asynchronous calls with AsyncOperator Although all data can be stored in the State, for historical reasons the user still needs to store some amount of data in an external KV storage location such as HBase. The user will then need to access this external data in the Flink Job, but since Flink has always been a single-threaded processing model the latency of accessing external data becomes a bottleneck for the entire system. While asynchronous access offers a direct means of solving this problem, it is nevertheless difficult for users to write multi-thread in UDF while ensuring exactly-once semantics. To solve this challenge, Alibaba proposed the AsyncOperator in Flink, which makes writing asynchronous calls in the Flink Job remarkably easy and offers a huge gain to Flink Job’s throughput. Enabling flexible task dispatch Flink’s key distinction is its design as a computing engine that unifies batches and streams, for which reason it is gaining interest among batch computing users who have experienced its lightning-fast stream computing capabilities. However, batch computing brings new challenges to the framework, motivating Alibaba’s extensive work to resolve these issues through original component contributions. In terms of task dispatch, Alibaba has provided a more flexible dispatch mechanism that can perform more efficient dispatches based on the dependencies between tasks. Beyond this, Flink’s native Shuffle Service is not binding with the TM, such that after a task is executed the original TM is retained and resources cannot be released, while the original Batch shuffle does not merge files (meaning it cannot be used in production). Alibaba has solved these problems while developing the Yarn Shuffle Service feature, discovering in the process that developing a new Shuffle Service involved inconvenient invasion of many locations in Flink’s code. To enable other developers to easily extend different Shuffles, Alibaba modified the Flink Shuffle architecture to make it pluggable. As an example of how it can be deployed, Alibaba’s search service currently uses Flink Batch Job to serve production. Looking Forward The optimizations described above have covered three years of intensive polishing by Alibaba, following which Blink has begun to work effectively for Alibaba’s unique needs as a large-scale production environment. Optimization and improvement of Runtime is endless, however, and Alibaba continues to work on a number of important improvements and optimizations bound for future versions of Flink. This article is part of Alibaba’s Flink series . Alibaba Tech First hand and in-depth information about Alibaba’s latest technology → Facebook: . Twitter: . “Alibaba Tech” “AlibabaTech”

Better to Give and to Receive: Alibaba’s Open-source Contributions to Flink

From version 1.1.0 to 1.6.0, Apache Flink’s relentless improvement exemplifies open-source development. This article is part of Alibaba’s Flink series . Open-source big data computing engine Apache Flink, or Flink for short, has gained popularity in recent years as a powerful framework for both batch processing and stream processing that can be used to create a number of event-based applications. Flink is first of all a pure stream computing engine with a data stream basic data model. A stream can be infinite and borderless, which describes stream processing in the general sense, or can be a finite stream with boundaries, as in the case of batch processing. Flink thus uses a single architecture to support both stream and batch processing. As conveyed by its slogan “Stateful Computations Over Streams”, Flink has the additional strength of supporting stateful computing. Processing is called stateless when the result of processing an event or piece of data is only related to the content of that event itself. Alternatively, if the result is related to previously processed events, it is called stateful processing. Some complicated data processing, such as aggregation, joins two streams that are both stateful processing. As well as relying on its support for massive data operations, Alibaba Group has contributed significantly to Flink’s development over the past several years, including with the addition of important components such as the Async I/O feature. Drawing on insights from Alibaba developer Chong (Jark) Wu, this article introduces key functions available in the framework’s current version before looking at the historical changes it has undergone to improve specific performance areas. Building Blocks of Apache Flink Flink’s increasing popularity is founded largely on four building block components, namely its Checkpoint, State, Time, and Window mechanisms. The Checkpoint mechanism is among Flink’s most important features. Flink implements distributed consistency snapshots based on the Chandy-Lamport algorithm, providing exactly-once semantics; by contrast, prior stream computing systems such as Strom and Samza did not effectively solve the exactly-once problem. After providing consistent semantics, Flink introduces a managed state and provides API interfaces that users can manage states with while programming, making programming as easy as using Java sets. Flink further implements the watermark mechanism that solves the data out-of-order and data late-arriving problems that occur in event-time based processing. Finally, because stream computing is generally based on windows, Flink provides a set of out-of-box window operations including tumbling, sliding, and session windows, and supports flexible custom windows to meet special requirements. Historical Changes to the Flink API Flink’s powerful framework has benefited from a number of groundbreaking API enhancements to each of its previous versions. During its 1.0.0 period, Flink introduced the State API, consisting of ValueState, ReducedState, ListState, and so on. The State API can be considered the biggest innovation of Flink. It allows users to use Flink’s managed state as Java Collections API and automatically enjoy state consistent guarantees without any loss of state due to failure. As a sign of its significance, even the subsequent Apache Beam State API borrowed extensively from this component. Flink’s 1.1.0 period saw the start of support for Session Window and correct processing for out-of-order and late-arriving data, assuring accuracy for final results. In the 1.2.0 period, Flink’s ProcessFunction API was provided, which is a lower-level API for implementing more advanced and complex functions. As well as being able to register various State types, it also supported registration timers (EventTime and ProcessingTime) and is often used to develop event-based, time-based applications. In the 1.3.0 period, Flink received the Side Output function. This is important because there is generally only one output type for the operator’s output, while another type may sometimes be needed. For example, as well as outputting main stream data, it may also be advantageous to output some amount of abnormal and late-arriving data in the form of side streams and then hand them to different downstream nodes for processing. Side Output specifically supports this kind of multiplexed output. Flink’s 1.5.0 phase saw the addition of BroadcastState, which is a State API extension that is used to store data broadcast from upstream. Because it is broadcast from the upstream, the BroadcastState data in each task of this operator is exactly the same. Based on this State, it becomes possible to more effectively resolve the dynamic rule functions in CEP and unequal join scenarios in SQL. Lastly, in its 1.6.0 period Flink received its State TTL and DataStream Interval Join functions. State TTL realized specification of time-to-live parameters (TTL) while applying for a certain state and ensured that this state can be specified for automatic cleanup by the system after a given time interval. Prior to this version, users have to register a timer within ProcessFunction and then manually clear the State using the timer’s callback. From 1.6.0 onward, the Flink framework solves this problem natively based on TTL. DataStream Interval Join makes interval Join possible. For example, a five-minute interval Join means that each record in the left stream joins right-stream data falling within five minutes before and five minutes after the joining. Historical Changes to the Flink High-Level API Parallel to the changes discussed in the previous section, Flink underwent a number of changes to its high-level API as it progressed through versions. In Flink’s 1.0.0 period, its Table API (or structured data processing API) and CEP (complex event processing API) frameworks were added to the library. The Table API is a structured, high-level API that supports the Java and Scala languages in a similar way to Spark’s DataFrame API. It is also highly similar to SQL. Table API and SQL are both relational APIs and can share lots of implementations. Therefore, in Flink 1.1.0, the community refactored the entire Table module based on Apache Calcite, which supported both the Table API and SQL and allowed the them to share most of the code. In the 1.2.0 period, the Flink community supported rich built-in window operations on the Table API and SQL, including Tumbling Window, Sliding Window, and Session Window. During the 1.3.0 period, the Flink community proposed the Dynamic Table concept. With it, stream and table are dual and can be transformed to each other without losing any information. This provides a foundation for the unification of stream and batch. At the core of implementing Dynamic Table is the Retraction mechanism, which alone allows multi-level Aggregate and multi-level Join to be implemented correctly, and which alone can ensure the correctness of semantics and results of streaming SQL. Additionally, in this version, Flink also supported the scalability of CEP operator — in other words, the parallelism of CEP operator can now be changed without loss of state. Finally, in Flink version 1.5.0, Join is supported on Table API and SQL, including regular stream Join and time-windowed stream Join. Additionally, SQL Client was added. The SQL Client provides an easy way of writing, debugging, and submitting table programs to a Flink cluster without a single line of Java or Scala code. The SQL Client CLI allows for retrieving and visualizing real-time results from the running distributed application on the command line. Historical Changes to Flink Checkpoint and Recovery Flink’s Checkpoint mechanism has been supported since its early stages and has since become a core feature, and the Flink community has also been working hard to improve the efficiency of checkpoint and recovery. Flink’s 1.0.0 period supported RocksDB statebackend. Prior to this version, all state data can only be stored in memory. Because JVM memory is fixed in size and a rise in the size of data will lead to the FullGC and OOM problems. This makes it difficult for Flink to be used stably in the production environment. To store more data or a larger state, the RocksDB statebackend is needed. RocksDB is a file-based embedded database that stores data to the disk while providing efficient read and write performance. Using RocksDB thus avoids the OOM problem. In its 1.1.0 period, Flink gained support for the asynchronization of RocksDB Snapshot. In previous versions RocksDB’s Snapshot process was synchronous, which blocks processing of the main data stream and affects throughput. With asynchronization support, Flink’s throughput has since been greatly improved. In the 1.2.0 period, support was added for the re-scalable KeyedState and OperatorState with the introduction of the KeyGroup mechanism. This is to say that the parallelism of operators can be changed without losing the state. In its 1.3.0 period, Flink added support for its Incremental Checkpoint mechanism, which indicated that Flink’s stream computation had absolutely reached a production-ready state. Incremental Checkpoint only stores and persists new incremental state snapshots. For general stream computing, gigabyte-level states and even terabyte-level states are quite common, and if the states for all data are stored to distributed storage each time then the cost to the network will become exceedingly high. In this version, Flink also supported fine-grained recovery features. Instead of restarting the complete ExecutionGraph in the event of a task failure, Flink was able to restart only the affected subgraph and thereby significantly decrease recovery time. Lastly, in the Flink 1.5.0 period, a mechanism was introduced for local state recovery. Because of the checkpoint mechanism, the State will be stored persistently in distributed storage. Taking HDFS, for instance, data will need to be re-downloaded from the remote HDFS when failover occurred. If the state is especially large, the download time will run long, and the time spent on failover recovery will likewise lengthen. The local state recovery mechanism backed up the state file locally in advance, so that when a job experiences failover it could then be restored locally instead of by downloading the state file from the remote HDFS, thus improving recovery efficiency. Historical Changes to Flink Runtime Notable developments in Flink Runtime begin from the version 1.2.0 period, when Alibaba directly contributed the high-profile Async I/O feature to the Flink community. Async I/O’s main purpose is to solve system bottlenecks caused by network latency when interacting with external systems, as for example when querying the external HBase table to fill some columns. A synchronous approach results in each query operation being blocked, and the data stream becomes stuck due to frequent I/O requests. Using asynchronous I/O, requests for N asynchronous queries can be initiated simultaneously without blocking the main data stream, improving the throughput of the entire job and utilization of the CPU. Flink version 1.3.0 introduced a HistoryServer component. HistoryServer allows users to query the status and statistics of completed jobs that have been archived by a JobManage, which is useful when doing in-depth troubleshooting after job is completed. In the Flink 1.4.0 period, an end-to-end exactly-once semantic guarantee was added. Exactly-once means that each record of input data will take effect on the final result exactly once even if software or hardware failures occur, thus ensuring data cannot be lost or computed twice. Prior to version 1.4.0, Flink provided the exactly-once guarantee within the Flink program, but did not include outputs to external systems, making it possible that duplicate data would be written to an external system if failover happened. At that time the general solution was to use an idempotent external database. In 1.4.0, Flink supported the end-to-end exactly-once semantic guarantee by the two-phase commit algorithm. It further supported Kafka’s end-to-end guarantee with a built-in approach and provided the TwoPhaseCommitSinkFunction for users to create their own exactly-once data sinks. During its 1.5.0 period, Flink released a new deployment model and processing model. The rewrite of Flink’s deployment and process model (internally known as FLIP-6) had been in the works for more than a year and was a substantial effort from the Flink community. It has been the most significant improvement of a Flink core component since the project’s inception. In a nutshell, the improvements add support for dynamic resource allocation and dynamic release of resources on YARN and Mesos schedulers for better resource utilization, failure recovery, and also dynamic scaling. Version 1.5.0 additionally refactored the network stack, for the reason that in the old version, communications between multiple upstream and downstream tasks shared a single TCP connection. When a task was back pressured, all tasks sharing its connection would be blocked. For Flink 1.5, the community worked on two efforts to improve Flink’s network stack: credit-based flow control and improved transfer latency. Credit-based flow control reduces the amount of data “on the wire” to a minimum while preserving high throughput. This significantly reduces the time to complete a checkpoint in back pressure situations. Moreover, Flink is now able to achieve much lower latencies without a reduction in throughput. Takeaways Among Flink’s greatest strengths is its unification for both stream and batch processing. Current development trends favor stream computing as the more mature leading model, and the approach has undergone verification in large-scale production environments at numerous companies. The Flink community will continue to improve the performance and functionality of stream computing in ways including the extension of a richer functionality for Flink SQL and the introduction of novel optimizations. Joining these efforts, the community will devote its efforts to enhancing its ecological capabilities, from batch processing to machine learning and beyond. (Original article by Wu Chong伍翀) This article is part of Alibaba’s Flink series . Alibaba Tech First hand and in-depth information about Alibaba’s latest technology → Facebook: . Twitter: . “Alibaba Tech” “AlibabaTech”

A Brief History of Flink: Tracing the Big Data Engine’s Open-source Development

This is the last part of a 5-part blog series. See , , , and . part 1 part 2 Part 3 Part 4 At their core, Serverless functions are about running pure business logic or ‘functions’ inside a container. These functions could be anything that makes sense to a business application. Some common examples include: Payment Processing Alerting — sending an alert to a Security Ops Console Stream Processing — enriching a message with customer information Doing a credit check Submitting an insurance claim Common use cases primed for Serverless functions include key industries, such as: Power & Utilities — where streaming data is intercepted and analyzed to perform demand forecasting, equipment reliability management, etc. Oil and Natural Gas Industry — Serverless is employed to perform log analytics to provide Single Views of Oil Wells Advertising — serving mobile & online users with instant & relevant offers based on their browsing history, location and buyer characteristics. Financial Services — such as banking and trading applications — where various facets of user data are populated in real-time in the customer portal. Healthcare — for example, papulation of health management data Let us examine some design patterns for common industry use cases to see how Serverless can help you accelerate innovation and software delivery for key event-driven applications. We’ll be using the open-source for illustrating these architectures, as it is the most flexible and is not locked to a specific cloud provider or related services, so that your apps can run anywhere. You can swap Fission with any serverless/cloud offering of your choice, too! Fission serverless framework Reference Architecture #1: Internet of Things (IoT) Typically in IoT — from industrial internet to wearables, to smart cars — a set of autonomous physical devices such as sensors, actuators, smart appliances and wearables, collect various types of data and communicate with an application running in the datacenter (either cloud/on-premises) using an IP protocol. Commonly, a lot of the data is aggregated using a gateway and then sent into a platform that can analyze all these variables for various business insights (performance, trends, triggered events, and more.) The overall flow in an IoT application can be orchestrated using Fissions: Data is aggregated using a Gateway and sent into to various message queues running on a cluster of Kafka servers running on Kubernetes managed pods Fission functions are invoked based on the overall pipeline flow: For a given file placed in a message queue, the contents of the file are passed into a Fission function that first normalizes it, extracts variables of interest and then sends the output into a NoSQL database or a file system. The second function will run in response to the normalized file being placed in the NoSQL database. It will read the contents of the file, perform computations as needed (based on the use case) and then invoke microservices that perform functionality such as sending the data into a Data Lake or a Data Mart for further analysis. Fission functions can be written for any kind of processing actions in the event stream. The FaaS scales up or down as needed in response to data volumes. Serverless Applications — Reference Architecture for IoT Use Cases Reference Architecture #2: Financial Services (Payments Processing, Risk Calculations, etc.) In the financial services industry, critical tasks such as payments processing, compliance checks and risk metrics can be calculated in real time using a Fission-based Serverless architecture. The overall flow in a financial application can be orchestrated using Fission: Developer deploys Fission functions as a shared capability across several applications that are the frontend to a variety of payment processing gateways. These applications handle user authentication, registration & collection of payment related data. These systems also interface on the backend with a variety of databases that record transaction data. Fission functions are created to parse a given input datastream that has the following variables: user’s credit card data, location of the transaction, any other demographic information, etc. The first function can call a fraud detection API and based on the results of the check persist the data into an in-memory datagrid. The second set of functions is invoked when the check has either passed or failed. If the check has passed, the function approves the payment and sends the user a confirmation. If the payment is suspected to be fraudulent, another function is called which alerts a Fraud detection system at the backend. Serverless Applications — Reference Architecture for Financial Services Use Cases ReferenceArchitecture #3: Web Applications Augmentation The vast majority of enterprise applications are three-tier web applications. These do everything from onboarding new customers, to interfacing with various back-office functions performing a range of business tasks as well as technology tasks (such as backups, reporting, alerting, data enrichment.) An emerging pattern is for Serverless Fission functions to handle these tasks without impacting the complicated development and deployment methodology of a monolithic n-tier application. The overall flow in web applications augmentation can be orchestrated using Fission: Business users continue to interface with the (legacy) application, but can take advantage of the augmented functionality provided by Fission functions Functions are created for a range of compute and data processing tasks as depicted The main application triggers these functions as it opportunistically needs to do Dedicated functions perform the appropriate logic as depicted, and if needed “report back: to the application. This could be seen as a way to gradually refactor or decompose/enhance existing legacy applications into microservices, or easily adding additional functionality that is not dependent on the development and release processes of the “mother ship”. Serverless Applications — Reference Architecture for Web Apps Use Cases Reference Architecture #4: Modernizing Legacy Systems Legacy systems such as mainframes, N-tier applications are prevalent and deeply rooted in industries such as Banking, Insurance, Retail, and more. Oftentimes it is not possible to simply rip and replace these systems (due to complexity, business dependencies, time it would take to re-architect, as well as the fact that often these legacy systems still perform quite well and have mature usage patterns.) To ensure interoperability with modern interfaces, such as web services and mobile applications, Fission functions can provide standards-based API wrappers around these systems. For instance: Requests are sent in to the API Gateway which performs basic transformation and invokes Fission functions as needed based on the endpoint being requested Functions take in the input data (e.g. JSON) and convert it to a format the end system can process and then calls the (legacy) business system Once the system responds, functions can take the response and convert it into a format the source system understands and then invoke the API Gateway with the appropriate response. Serverless Applications — Reference Architecture for Modernizing Legacy Applications Reference Architecture #5: Machine Learning and Deep Learning with Apache Spark Across-industries, we see applications increasingly infusing business processes and big data with Machine Learning (ML) capabilities. From fraud detection, customer behavior and pipeline analysis, virtual reality, conversational interfaces, chatbots, consumption trends, video/facial recognition, and more — it seems that ML (and AI) is everywhere. For most companies, ML and Predictive Analytics initiatives have typically been silo’d to a specific project within the organization. To realize the real value of ML, it’s advantageous for the data, learnings, algorithms and models to be shared across applications. Fission, in conjunction with technologies such as Apache Spark, enables you to provide the data of stream processing and trend predictions to be consumed by a variety of end users/applications. The overall flow in Machine Learning applications can be orchestrated using Fission and Apache Spark: Data from business operations is ingested in real-time into a cluster of Kafka-based message queues. Spark Streaming is used to ingest this data in micro-batches, typically based on a time window. This data is stored into a data lake for batch analysis as well as sent into a Spark MlLib runtime where different predictive models are stored. These models are based on general-purpose ML algorithms that run on Spark. These include both supervised and unsupervised algorithms — e.g. clustering, classification algorithms, etc. Once the results of the model are written into a NoSQL database or an in-memory datagrid, Fission functions are triggered. These functions do a range of business-critical functionality. For example: updating business analytics dashboards, sending real-time customer offers & alerting customer agents. Serverless Applications — Reference Architecture for Machine Learning and Big Data Applications Conclusion Serverless has immense industry mindshare and is emerging as the hottest trend after Kubernetes. However, as with other transformational technologies, significant concerns remain around lock-in, scalability and ecosystem integration. help industry verticals solve their pressing business challenges using a highly flexible, state of the art, and standards-based FaaS platform that can run anywhere- on premises and on any public cloud. Open source Serverless solutions

What is Serverless — Part 5: Reference Architectures for Common Types of Serverless Applications

This article was written by Marcin Niczyporuk , the director and head of Cognitive Services at intive . intive is a software company focused on digital product development with more than 18 years of experience and 150+ apps. Many say 2018 will be the year AI finally hits the mainstream, helping organizations and employees unleash their real potential. However, look around, and you’ll realize many companies have already taken the first big leaps. Thanks to exponential technologies such as , mobile solutions and the cloud, companies around the world have the resources to invest in AI solutions and augment their processes. In fact, , 54 percent of businesses claim AI has already increased productivity within their organizations. machine learning according to PwC However, for those who have yet to take the much-needed jump into the AI world, the idea of adopting the can be, frankly, daunting. Especially considering the amount of data required for AI to be impactful. But finding this data is not as complicated as you might believe. Here’s how to implement AI in your product — even if you don’t think you collect enough data in the first place: technology Check to see if there are existing datasets at your disposal Tools that leverage recommendation technology, natural language processing, image recognition or deep learning algorithms undoubtedly require a vast amount of data to train machine learning models. However, when compiling data, it’s not always necessary to start from scratch. In the early stage of product and service development, it’s important for companies to verify what out-of-the-box data services already exist on the market. Tech giants like Amazon, Microsoft, Google, IBM and Oracle all boast PaaS cloud offerings, which feature existing public data sets and pre-built AI services for companies to leverage. Dig a little deeper, and there’s actually a whole range of free public data sources available in unsuspecting places — like or the , for example. In fact, this lists 30 places companies can begin the search for data. World Bank Open Data Bureau of Justice Forbes article There’s also a handful of organizations that sell data for this very purpose. Companies such as , , and are data market platforms that have already done the hard work of collecting the data and are now trading it through an API. Other organizations like and actually crowdsource data and data science services — so if a company needs something in particular, it can be created for them. DAVEX Quandl DatastreamX Qlick Data Market Kaggle Drivendata However, remember: these public data sources might not work if you have a genuinely disruptive model — we’re talking something like , which claims to be the first GDPR compliant voice assistant, as it doesn’t rely on the cloud. So in cases like these, a company might need to put significantly more work into getting the data they need. Snips.ai Your back office software could be a ‘treasure island’ Ask any company that hasn’t formally invested in AI if they have valuable data on customer behavior, and they’ll likely say no chance. The thing is, however, many companies don’t realize they’re collecting this data through everyday processes — which in turn, can be used to develop powerful AI tools. Customer emails can provide a huge wealth of data; for example, they can be examined to extract information on how the customer’s ‘mood’ changes over time. Invoicing systems can provide great data, too. This year, for instance, Canadian accounting software company how polite citizens are by extracting data from business invoices. So can CRM notes, customer survey responses, meeting minutes, chat transcripts, and more. Freshbooks measured Now, imagine being able to measure 10 years of email and invoice history, for example, . A company could combine the data to predict what actions a customer might take once they start communicating a certain way. If the AI predicts a customer is not going to pay, for example, it would enable the company to react quickly — and maybe even save the contract. together So how can a company find out whether their once overlooked data is treasure or mere fool’s gold? Find a vendor. A vendor will consult companies on their current data structure processes, and explore ways in which the data can be extracted — think extracting data from text, for example. They’ll also present the data to a company in a digestible way, through modern infrastructures like , , or . Hadoop Hortonworks Tableau Power BI A good vendor will have experience working with a range of client domains, as this diversity enables them to learn quickly. However being specialized in certain vertical is also valuable, so try to find a vendor which clearly markets what they’re best at. And finally, a good vendor should be able to offer packaged data science solutions — such as pre-built models, frameworks, and reusable components from previous projects — to speed up the entire of process of ‘mining for data.’ Start logging If a company has already checked out public data sets to no avail, and the back office software data doesn’t quite meet expectations — it’s time to start logging. This means starting to collect literally everything; of course, with an experienced vendor’s help. Yes, companies should look to start compiling CRM, chat and invoicing data etc. However, they should also look to public datasets such as from social media, their real-time business insights from places like Twitter for sentiment analysis, and even the weather. This will all go into a data warehouse. You never know what you might need the data for — that’s why it’s important to log everything in sight. Keep in mind, though, that creating a data warehouse is no trivial task. That said, it’s not particularly easy to find experts with strong analytical skills on the market. So when searching for an expert vendor, be sure to request case studies, testimonials from previous clients, or proof of concept tests to ensure the data scientists can really collect the data efficiently, and help produce valuable insight from it. Many companies new to AI might feel apprehensive to take the first leap. However, if companies know where to look for help — whether it be to public datasets, existing customer information, or innovative data scientists — the whole process doesn’t seem as daunting. And even if it does, it’s prudent for companies to pull themselves up by their bootstraps, anyway. Because if they don’t invest in AI sooner or later, they’ll most certainly be left behind.

How To Implement AI Into Your Product If You Are Taking The First Leap

February 12, 2018 Javascript https://medium.com/@baphemot/whats-new-in-react-16-3-d2c9b7b6193b https://medium.com/dailyjs/reacts-%EF%B8%8F-new-context-api-70c9fe01596b https://blog.hackages.io/https-blog-hackages-io-evanyoubhack2017-cc5559806157 https://medium.com/welldone-software/an-overview-of-javascript-testing-in-2018-f68950900bc3 https://blog.graph.cool/prisma-cloud-preview-invite-only-c251bebf5670 https://divu.in/prisma-deep-dive-3162dea2820c https://medium.com/@nate_wang/introducing-rekit-studio-a-real-ide-for-react-and-redux-development-baf0c99cb542 https://medium.freecodecamp.org/how-to-build-a-real-time-graph-using-javascript-pusher-d15ccb7a4b82 https://blog.angular.io/run-angular-cli-repos-directly-in-your-browser-41332fd80901 https://codeburst.io/javascript-es6-the-spread-syntax-f5c35525f754 https://developers.caffeina.com/chiccocoin-learn-what-is-a-blockchain-by-creating-one-in-nodejs-12929a89208b https://medium.mybridge.co/javascript-top-10-articles-for-the-past-month-v-feb-2018-cb8a4949494f https://medium.freecodecamp.org/evolving-patterns-in-react-116140e5fe8f https://medium.muz.li/7-easy-steps-for-designers-to-create-an-app-in-react-f970114e886b https://medium.freecodecamp.org/learn-to-spot-red-flags-in-your-react-javascript-code-d52d5fac85f4 https://medium.freecodecamp.org/service-workers-the-little-heroes-behind-progressive-web-apps-431cc22d0f16 https://medium.com/@hanilim/http-codes-as-valentines-day-comics-8c03c805faa0 https://medium.freecodecamp.org/five-common-problems-in-graphql-apps-and-how-to-fix-them-ac74d37a293c DevOps https://medium.com/@johnbarton/developers-on-call-68c94e6a762a https://medium.com/@copyconstruct/on-call-b0bd8c5ea4e0 https://blog.hasura.io/an-exhaustive-guide-to-writing-dockerfiles-for-node-js-web-apps-bbee6bd2f3c4 https://blog.hasura.io/the-ultimate-guide-to-writing-dockerfiles-for-go-web-apps-336efad7012c https://medium.com/seldon-open-source-machine-learning/introducing-seldon-core-machine-learning-deployment-for-kubernetes-e10e94c19fd8 Cloud & Serverless https://dbamantra.com/top-5-cloud-computing-certifications-of-2018/ https://towardsdatascience.com/go-serverless-with-firebase-5348dedb70e9 https://medium.com/containers-on-aws/scaling-a-realtime-chat-app-on-aws-using-socket-io-redis-and-aws-fargate-4ed63fb1b681 AI http://bair.berkeley.edu/blog/2018/02/06/phri/ http://introtodeeplearning.com/ http://news.mit.edu/2018/alumni-call-mit-champion-artificial-intelligence-public-good-0209 https://cbmm.mit.edu/publications/object-oriented-deep-learning https://medium.com/netflix-techblog/ava-the-art-and-science-of-image-discovery-at-netflix-a442f163af6 https://medium.com/intuitionmachine/neurons-are-more-complex-than-what-we-have-imagined-b3dd00a1dcd3 https://towardsdatascience.com/building-a-deep-neural-net-in-google-sheets-49cdaf466da0 https://medium.freecodecamp.org/how-you-can-train-an-ai-to-convert-your-design-mockups-into-html-and-css-cc7afd82fed4 https://towardsdatascience.com/the-5-clustering-algorithms-data-scientists-need-to-know-a36d136ef68 https://medium.freecodecamp.org/using-machine-learning-to-predict-the-quality-of-wines-9e2e13d7480d https://medium.com/init27-labs/tackling-adversarial-examples-introspective-cnn-b28c67176575 https://medium.com/@ment_at/datastream-io-open-source-anomaly-detection-64db282735e0 https://medium.com/airbnb-engineering/fighting-financial-fraud-with-targeted-friction-82d950d8900e https://medium.com/meta-design-ideas/exploring-python-ecosystem-for-machine-learning-things-you-should-know-2a1ca32c8409 https://medium.freecodecamp.org/https-medium-freecodecamp-org-how-to-predict-rent-and-select-the-best-lease-duration-to-save-money-5cf35145d398 Big Data https://tech.flipkart.com/overview-of-flipkart-data-platform-20c6d3e9a196 https://medium.com/google-cloud/learning-to-analyze-huge-bigquery-datasets-using-python-on-kaggle-2c6c6153f542 https://medium.com/@bongban02111/big-data-revolution-e22e76bd0038 Blockchain https://medium.com/@phoneum/what-is-phoneum-7eb9226bbe64 https://shitcoin.com/iota-cannot-be-used-for-iot-loss-of-funds-may-occur-e45b1ed9dd6b https://hackernoon.com/a-surprising-way-to-win-in-a-crashing-crypto-market-7515fbbc85f https://medium.com/crypto-oracle/the-single-biggest-misconception-of-crypto-is-3d2625a08405 https://medium.com/@lopp/satoshi-roundtable-iv-recap-d7828e9f9e5e https://medium.com/@silverlinings171/the-beauty-of-nebulas-nas-16906f47dce9 https://medium.com/@ternio/ternio-100k-airdrop-8dbd49e28e52 https://hackernoon.com/forget-bitcoin-diversify-with-these-four-cryptocurrencies-now-7f2bd6a491e0 https://hackernoon.com/best-sleeper-coins-for-2018-71a20e16e61b https://medium.com/consensus-ai/consensus-ai-world-citizen-airdrop-update-d1a7046fd29 Personal Growth https://journal.thriveglobal.com/want-to-become-a-multi-millionaire-do-these-15-things-immediately-e1e779a6978f https://medium.com/personal-growth/money-doesnt-make-you-happy-but-this-does-11f4c7803da4 https://medium.com/swlh/my-thoughts-on-money-after-being-both-rich-and-poor-634c75c93734 https://medium.com/swlh/41-things-you-should-say-no-to-to-become-the-person-you-want-to-be-in-life-and-business-b5c7832c3b36 https://medium.com/the-mission/how-to-design-a-lifestyle-that-eliminates-distraction-and-enables-hyper-focus-c1f350d5b5ba https://medium.com/personal-growth/the-7-best-pieces-of-advice-for-living-a-happy-life-29fd55b011fb https://medium.com/the-mission/why-you-should-stop-expecting-others-to-be-who-theyre-not-68e2d781da07 https://medium.com/swlh/i-work-for-free-458c7fdcac9e https://medium.com/the-mission/want-to-support-your-grieving-friend-5-truths-about-what-really-helps-f478a04f611a https://medium.com/personal-growth/the-dark-emotional-side-of-minimalism-48fa9a0488a8 https://medium.com/personal-growth/9-stoic-practices-that-will-help-you-thrive-in-the-madness-of-modernity-af9f1bdd56f4 https://medium.com/personal-growth/13-keys-to-success-you-can-learn-from-the-monk-who-sold-his-ferrari-440cb280a699 https://journal.thriveglobal.com/you-make-or-break-your-life-between-5-7-am-70e2717f9e67 https://journal.thriveglobal.com/decisions-determine-destiny-heres-how-to-make-the-best-one-s-690a035b00c4 Startup https://medium.com/swlh/want-to-build-a-successful-startup-adopt-these-3-mindsets-immediately-9a2840d78525 https://m.signalvnoise.com/the-world-needs-more-modest-linear-growth-companies-please-make-some-609b5a10a9e0 https://medium.com/swlh/how-focusing-on-the-customer-not-the-competition-brought-us-over-1-million-new-signups-in-a-year-8fbcc8c01bbe https://hackernoon.com/why-you-should-build-minimum-delight-able-product-mdp-a2d27897b5a4 Misc https://medium.com/@fluperofficial/6-best-alternates-to-google-and-apple-app-stores-bae941a8b8e9 https://medium.com/coderbyte/the-10-best-coding-challenge-websites-for-2018-12b57645b654 https://www.edsurge.com/news/2018-01-22-five-ways-teachers-can-use-and-create-augmented-reality-experiences

Week 6: Reading list & news to keep up-to-date with Javascript, Python, DevOps, Cloud, Serverless…

In the summer of 2023, a cheeky blog post titled "Kafka is dead, long live Kafka" came across my feed. It was written by a company I hadn’t heard of yet, WarpStream. It’s a long blog post, but I couldn’t stop reading it. What was being described was incredibly interesting. What WarpStream was proposing and had built is an Apache Kafka® protocol compatible data streaming platform built directly on top of S3. It's delivered as a single, stateless Go binary, so there are no local disks to manage, no brokers to rebalance, and no ZooKeeper to operate. If you aren’t in the know about Kafka, it is an open-source event streaming platform that can collect, store, process, and publish streaming data in real time. Kafka was originally developed at LinkedIn and open-sourced in 2011. It is written in Java and Scala and is in wide deployment, but it is a large, complex, and difficult project to launch and manage. The Kafka API has been established and is supported by other players in this space, which makes for a robust community and exciting developments. Let’s Dive In Originally designed for LinkedIn’s data centers, Kafka's migration to the cloud has presented significant hurdles. The replication strategy often results in higher inter-availability zone bandwidth costs, and its management typically necessitates a dedicated team. This is the precise problem that WarpStream aims to address with its innovative solution. As previously mentioned, it is a Kafka protocol-compatible data streaming platform that runs directly on top of any commodity object store. Name the cloud, and it appears they run on it. I’m personally a big fan of the Go language, so I liked that it was developed in Go. What’s next? Using WarpStream I’m doing this in Ubuntu running in WSL on Windows 11. The docs give you a few options for installing the WarpStream agent; I used the one-line installation script and note, that you don’t need to sign up for anything to do this: curl https://console.warpstream.com/install.sh | bash With that installed, then I issued the warpstream demo command. This will do the following: Automatically sign you up for a temporary account that is valid for 12 hours.
Run an in-memory producer that will produce small JSON documents to a stream periodically.
Run an in-memory WebAssembly consumer that consumes the JSON documents and prints them to the standard console. This is what it looked like. The bottom line in green is the URL that I pasted into my web browser, which I’ll get to in a moment. After more than a screenful of feedback, I get the suggestion in blue to execute that comment in another terminal, and then we see the data getting produced to a topic and 4 different partitions. \With that all running, I pasted the URL into my browser and I saw this immediately. We see two agents running in the cluster overview Curious about the activity, I decided to delve into the topics. There, I found our topic with its four partitions, all showing write activity from the demo command. This whole demo is ephemeral, so you can try it out; it shuts itself down after an hour, so don’t try doing anything serious. There are a number of other features in there that I tried out; they’ve definitely bundled a lot of ease of use into the interface. If I had a nit to pick, it’s the UI for pop-up dialogs for creating and deleting objects; they have no ‘cancel’ option, and you have to hit the left navigation to bail out. There is a very cool startup wizard when you start building either a Serverless or BYOC cluster. It creates a temporary virtual cluster for a tutorial that then walks you through each step of getting started. The function seems to trigger if it is your first time creating a server in your account. Summary There are a lot of cool features in the WarpStream product that make it easy to set up and use, crazy easy. If you want a deep dive, I suggest their blog post that I linked in the intro, and for you to try it yourself. The fact that you can drop it into an existing Kafka setup and have it basically “just work”, is very compelling. If I had a nit to pick, it’s that there isn’t an open-source component, but on the other hand, it’s a product on top of an open protocol, which is basically how a lot of open-source companies monetize. They are currently sponsoring Benthos, which was recently integrated into WarpStream to give you integrated pipeline editing and management. That’s a really nifty integration that is worth its own blog. So, what the heck is WarpStream? It’s a really powerful, easy-to-use, Kafka API-compatible replacement for Apache Kafka. It should lower your operational overhead for running Kafka and make it cheaper to run your streaming workloads. If you’re streaming with Kafka or one of the other protocols, it’s worth checking out. **Check out my other What the Heck is… articles at the links below: What The Heck Is DuckDB?
What the Heck Is Malloy?
What the Heck is PRQL?
What the Heck is GlareDB?
What the Heck is SeaTunnel?
What the Heck is LanceDB?
What the heck is SDF?
What the Heck is Paimon?
What the Heck is Proton?
What the Heck is PuppyGraph?
What the Heck is GPTScript? ** In the summer of 2023, a cheeky blog post titled " Kafka is dead, long live Kafka " came across my feed. It was written by a company I hadn’t heard of yet, WarpStream. It’s a long blog post, but I couldn’t stop reading it. What was being described was incredibly interesting. What WarpStream was proposing and had built is an Apache Kafka ® protocol compatible data streaming platform built directly on top of S3. It's delivered as a single, stateless Go binary, so there are no local disks to manage, no brokers to rebalance, and no ZooKeeper to operate. If you aren’t in the know about Kafka, it is an open-source event streaming platform that can collect, store, process, and publish streaming data in real time. Kafka is dead, long live Kafka Kafka is dead, long live Kafka WarpStream Apache Kafka Kafka was originally developed at LinkedIn and open-sourced in 2011. It is written in Java and Scala and is in wide deployment, but it is a large, complex, and difficult project to launch and manage. The Kafka API has been established and is supported by other players in this space, which makes for a robust community and exciting developments. Let’s Dive In Let’s Dive In Originally designed for LinkedIn’s data centers, Kafka's migration to the cloud has presented significant hurdles. The replication strategy often results in higher inter-availability zone bandwidth costs, and its management typically necessitates a dedicated team. This is the precise problem that WarpStream aims to address with its innovative solution. As previously mentioned, it is a Kafka protocol-compatible data streaming platform that runs directly on top of any commodity object store. Name the cloud, and it appears they run on it. I’m personally a big fan of the Go language, so I liked that it was developed in Go. What’s next? Using WarpStream I’m doing this in Ubuntu running in WSL on Windows 11. The docs give you a few options for installing the WarpStream agent; I used the one-line installation script and note, that you don’t need to sign up for anything to do this: docs docs curl https://console.warpstream.com/install.sh | bash curl https://console.warpstream.com/install.sh | bash curl https://console.warpstream.com/install.sh | bash curl https://console.warpstream.com/install.sh | bash curl https://console.warpstream.com/install.sh | bash With that installed, then I issued the warpstream demo command. This will do the following: warpstream demo Automatically sign you up for a temporary account that is valid for 12 hours. Run an in-memory producer that will produce small JSON documents to a stream periodically. Run an in-memory WebAssembly consumer that consumes the JSON documents and prints them to the standard console. Automatically sign you up for a temporary account that is valid for 12 hours. Run an in-memory producer that will produce small JSON documents to a stream periodically. Run an in-memory WebAssembly consumer that consumes the JSON documents and prints them to the standard console. This is what it looked like. The bottom line in green is the URL that I pasted into my web browser, which I’ll get to in a moment. After more than a screenful of feedback, I get the suggestion in blue to execute that comment in another terminal, and then we see the data getting produced to a topic and 4 different partitions. \With that all running, I pasted the URL into my browser and I saw this immediately. We see two agents running in the cluster overview Curious about the activity, I decided to delve into the topics. There, I found our topic with its four partitions, all showing write activity from the demo command. This whole demo is ephemeral, so you can try it out; it shuts itself down after an hour, so don’t try doing anything serious. There are a number of other features in there that I tried out; they’ve definitely bundled a lot of ease of use into the interface. If I had a nit to pick, it’s the UI for pop-up dialogs for creating and deleting objects; they have no ‘cancel’ option, and you have to hit the left navigation to bail out. There is a very cool startup wizard when you start building either a Serverless or BYOC cluster. It creates a temporary virtual cluster for a tutorial that then walks you through each step of getting started. The function seems to trigger if it is your first time creating a server in your account. Summary There are a lot of cool features in the WarpStream product that make it easy to set up and use, crazy easy. If you want a deep dive, I suggest their blog post that I linked in the intro, and for you to try it yourself. The fact that you can drop it into an existing Kafka setup and have it basically “just work”, is very compelling. If I had a nit to pick, it’s that there isn’t an open-source component, but on the other hand, it’s a product on top of an open protocol, which is basically how a lot of open-source companies monetize. They are currently sponsoring Benthos , which was recently integrated into WarpStream to give you integrated pipeline editing and management. That’s a really nifty integration that is worth its own blog. Benthos Benthos So, what the heck is WarpStream? It’s a really powerful, easy-to-use, Kafka API-compatible replacement for Apache Kafka. It should lower your operational overhead for running Kafka and make it cheaper to run your streaming workloads. If you’re streaming with Kafka or one of the other protocols, it’s worth checking out. **Check out my other What the Heck is… articles at the links below: What The Heck Is DuckDB? What the Heck Is Malloy? What the Heck is PRQL? What the Heck is GlareDB? What the Heck is SeaTunnel? What the Heck is LanceDB? What the heck is SDF? What the Heck is Paimon? What the Heck is Proton? What the Heck is PuppyGraph? What the Heck is GPTScript? What The Heck Is DuckDB? What The Heck Is DuckDB? What The Heck Is DuckDB? What the Heck Is Malloy? What the Heck Is Malloy? What the Heck Is Malloy? What the Heck is PRQL? What the Heck is PRQL? What the Heck is PRQL? What the Heck is GlareDB? What the Heck is GlareDB? What the Heck is GlareDB? What the Heck is SeaTunnel? What the Heck is SeaTunnel? What the Heck is SeaTunnel? What the Heck is LanceDB? What the Heck is LanceDB? What the Heck is LanceDB? What the heck is SDF? What the heck is SDF? What the heck is SDF? What the Heck is Paimon? What the Heck is Paimon? What the Heck is Paimon? What the Heck is Proton? What the Heck is Proton? What the Heck is Proton? What the Heck is PuppyGraph? What the Heck is PuppyGraph? What the Heck is PuppyGraph? What the Heck is GPTScript? What the Heck is GPTScript? What the Heck is GPTScript? **

What The Heck is WarpStream?

Setting up Data Streams Using NiPyAPI: UI-Free Guide

In this blog post, we will be covering how you can combine and leverage the open-source streaming solution, , with , to improve the quality of your streaming flows. Buckle up! bytewax ydata-profiling Stream processing enables real-time analysis of data in-flight and before storage and can be or . stateful stateless is used for recommendations, pattern detection, or complex event processing, where the history of what has happened is required for the processing (windows, joining by a key, etc.). Stateful stream processing real-time is used for transformation that doesn’t require knowledge of other data points in the stream like masking an email or converting a type. Stateless stream processing inline Overall, data streams are widely used in the industry and can be found applied to use cases such as , , or . fraud detection patient monitoring event predictive maintenance One Crucial Aspect That All Data Streams Must Consider Is the Quality of the Data Unlike traditional models where data quality is usually assessed during the creation of the data warehouse or dashboard solution, . streaming data requires continuous monitoring It is essential to maintain data quality throughout the entire process, from collection to feeding downstream applications. After all, the cost of bad data quality can be high for organizations: “The cost of bad data is an astonishing 15% to 25% of revenue for most companies. (…) Two-thirds of these costs can be eliminated by getting in front on data quality.” — Thomas C. Redman, author of “Getting in Front on Data Quality” Throughout this article, we will show you how you can combine with to profile and improve the quality of your streaming flows! bytewa ydata-profiling Stream Processing for Data Professionals With Bytewax is an OSS stream processing framework designed specifically for Python developers. Bytewax It allows users to with capabilities similar to Flink, Spark, and Kafka Streams while providing a friendly and familiar interface and build streaming data pipelines and real-time applications 100% compatibility with the Python ecosystem. Using built-in or existing Python libraries, (Kafka, RedPanda, WebSocket, etc.) and out to various downstream systems (Kafka, parquet files, data lakes, etc.). connectors you can connect to real-time and streaming data sources write transformed data For the transformations, Bytewax with , , and methods and comes with familiar features such as recovery and scalability. facilitates stateful and stateless transformations map windowing aggregation Bytewax and is p . facilitates a Python-first and data-centric experience to data streams urposely built for data engineers and data scientists It allows users to and create customizations necessary to meet their needs without having to learn and maintain JVM-based streaming platforms like Spark or Flink. build streaming data pipelines and real-time applications Bytewax is well suited for many use cases, namely, , , , and more. Embedding Pipelines For Generative AI Handling Missing Values in Data Streams Using Language Models in a Streaming Context to Understand Financial Markets For use case inspiration and more information like documentation, tutorials, and guides, feel free to check . the bytewax website Why Data Profiling for Data Streams? and refers to the step of : its structure, behavior, and quality. Data Profiling is key to a successful start of any machine learning task thoroughly understanding our data In a nutshell, involves analyzing aspects related to the data’s format and basic descriptors (e.g., number of samples, number/types of features, duplicate values), its (such as the presence of missing data or imbalanced features), and other complicating factors that may arise during data collection or processing (e.g., erroneous values or inconsistent features). data profiling intrinsic characteristics , where circumstances might change fast and may require immediate action (e.g., healthcare monitoring, stock values, air quality policies). Ensuring high data quality standards is crucial for all domains and organizations, but is especially relevant for domains operating with domains outputting continuous data For many domains, data profiling is used from an exploratory data analysis lens, considering historical data stored in databases. On the contrary, for data streams, , where data needs to be checked at different time frames or stages of the process. data profiling becomes essential for validation and quality control continuously along the stream By embedding an , we can on the current state of our data and be alerted for any potentially critical issues — whether they are related to (e.g., corrupted values or changing formats), or to (e.g., data drifts, deviation from business rules and outcomes). automated profiling into our data flows immediately get feedback data consistency and integrity events happening in short periods of time In real-world domains — — automated profiling might save us from multiple brain puzzles and systems needing to be taken out of production! where you just know Murphy’s law is bound to strike and “everything can definitely go wrong” In what concerns data profiling, has consistently been a , either for or data. And no wonder why — ydata-profiling crowd favorite tabular time-series it’s one line of code for an extensive set of analysis and insights. Complex and time-draining operations are done under the hood: ydata-profiling and depending on the feature types (either numeric or categorical), it that are shown in the profiling report. automatically detects the feature types comprised in the data, adjusts the summary statistics and visualizations Fostering a , the package also , focusing on their pairwise and , and provides a thorough , from or values to and features. data-centric analysis highlights the existing relationships between features interactions correlations evaluation of data quality alerts duplicate constant skewed imbalanced It’s really a — with minimal effort. 360º view of the quality of our data Putting It All Together: bytewax and ydata-profiling Before starting the project, we need to first set our Python dependencies and configure our data source. First, let’s install the and packages ( bytewax ydata-profiling You might want to use a virtual environment for this — if you need some extra guidance!) check these instructions pip install bytewax==0.16.2 ydata-profiling==4.3.1 Then, we’ll upload the (License — CC0: Public Domain), which contains several measurements of from different IoT devices: Environmental Sensor Telemetry Dataset temperature, humidity, carbon monoxide liquid petroleum gas, smoke, light, and motion wget https://raw.githubusercontent.com/bytewax/air-quality-sensor/main/data/iot_telemetry_data_1000 , and the input would look like what we expect in a streaming platform . In this article, and create a dataflow using bytewax. In a production environment, these measurements would be continuously generated by each device such as Kafka to simulate the context we would find with streaming data, we will read the data from the CSV file one line at a time (As a quick side note, a dataflow is essentially a data pipeline that can be described as a directed acyclic graph — DAG) First, let’s make some : necessary imports from datetime import datetime, timedelta, timezone

from bytewax.dataflow import Dataflow
from bytewax.connectors.stdio import StdOutput
from bytewax.connectors.files import CSVInput
from bytewax.testing import run_main Then, we define our dataflow object. Afterward, we will use a stateless map method where we pass in a function to convert the string to a DateTime object and restructure the data to the format (device_id, data). The map method will make the change to each data point in a stateless way. The reason we have modified the shape of our data is so that we can easily group the data in the next steps to profile data for each device separately rather than for all of the devices simultaneously. flow = Dataflow()
flow.input("simulated_stream", CSVInput("/content/iot_telemetry_data_1000"))

# parse timestamp
def parse_time(reading_data):
    reading_data["ts"] = datetime.fromtimestamp(float(reading_data["ts"]), timezone.utc)
    return reading_data

flow.map(parse_time)


# remap format to tuple (device_id, reading_data)
flow.map(lambda reading_data: (reading_data['device'], reading_data)) Now, we will take advantage of the stateful capabilities of to gather data for each device over the duration of time that we have defined. expects a snapshot of the data over time, which makes the window operator the perfect method to use to do this. bytewax ydata-profiling In , we are able to produce summarizing statistics for a dataframe that is specified for a particular context. For instance, in our example, we can produce snapshots of data referring to each IoT device or to particular time frames: ydata-profiling from bytewax.window import EventClockConfig, TumblingWindow

# This is the accumulator function, and outputs a list of readings
def acc_values(acc, reading):
    acc.append(reading)
    return acc


# This function instructs the event clock on how to retrieve the
# event's datetime from the input.
def get_time(reading):
    return reading["ts"]

  
# Configure the `fold_window` operator to use the event time.
cc = EventClockConfig(get_time, wait_for_system_duration=timedelta(seconds=30))

# And a tumbling window
align_to = datetime(2020, 1, 1, tzinfo=timezone.utc)
wc = TumblingWindow(align_to=align_to, length=timedelta(hours=1))

flow.fold_window("running_average", cc, wc, list, acc_values)

flow.inspect(print) After the snapshots are defined, leveraging is as simple as calling the for each of the dataframes we would like to analyze: ydata-profiling ProfileReport import pandas as pd
from ydata_profiling import ProfileReport

def profile(device_id__readings):
    print(device_id__readings)
    device_id, readings = device_id__readings
    start_time = readings[0]['ts'].replace(minute=0, second=0, microsecond=0).strftime('%Y-%m-%d %H:%M:%S')
    df = pd.DataFrame(readings)
    profile = ProfileReport(
        df,
        tsmode=True,
        sortby="ts",
        title=f"Sensor Readings - device: {device_id}"
    )

    profile.to_file(f"Ts_Profile_{device_id}-{start_time}.html")
    return f"device {device_id} profiled at hour {start_time}"


flow.map(profile) In this example, we are writing the images out to local files as part of a function in a map method. These could be reported out via a messaging tool or we could save them to some remote storage in the future. Once the profile is complete, the dataflow expects some output so we can use the built-in to print the device that was profiled and the time it was profiled at that was passed out of the profile function in the map step: StdOutput flow.output("out", StdOutput()) There are multiple ways to execute Bytewax dataflows. In this example, we use the same local machine, but Bytewax can also run on multiple Python processes, across multiple hosts, in a , using a , and . Docker container Kubernetes cluster more In this article, we’ll continue with a local setup, but we encourage you to check our helper tool which manages Kubernetes dataflow deployments once your pipeline is ready to transition to production. waxctl Assuming we are in the same directory as the file with the dataflow definition, we can run it using: python -m bytewax.run ydata-profiling-streaming:flow We can then use the profiling reports to validate the data quality, check for changes in schemas or data formats, and . compare the data characteristics between different devices or time windows In fact, we can leverage the that highlights the differences between two data profiles in a straightforward manner, making it easier for us to detect important patterns that need to be investigated or issues that have to be addressed: comparison report functionality snapshot_a_report = ProfileReport(df_a, title="Snapshot A")
snapshot_b_report = ProfileReport(df_b, title="Snapshot B")

comparison_report =snapshot_a_report(snapshot_b_report)
comparison_report.to_file("comparison_report.html") Ready to Explore Your Own Data Streams? Validating data streams is crucial to identify issues in data quality in a continuous manner and compare the state of data across distinct periods of time. For organizations in , , , and — all working with continuous streams of data — a , from quality assessment to data privacy. healthcare energy manufacturing entertainment utomated profiling is key to establishing data governance best practices This requires the analysis of snapshots of data which, as showcased in this article, can be achieved in a seamless way by combining and . bytewax ydata-profiling takes care of all the processes necessary to handle and structure data streams into snapshots, which can then be summarized and compared with through a comprehensive report of data characteristics. Bytewax ydata-profiling Being able to appropriately process and profile incoming data opens up a plethora of use cases across different domains, from the to the highlighting and mitigation of additional issues that derive from real-world activities, such as (e.g., fraud or intrusion/threats detection), , and other events that deviate from the expectations (e.g., data drifts or misalignment with business rules). correction of errors in data schemas and formats anomaly detection equipment malfunction Now, you’re all set to start exploring your data streams! Let us know what other use cases you find, and as always, feel free to drop us a line in the comments, or find us at the for further questions and suggestions! Data-Centric AI Community See you there! Acknowledgments ) and Oli Makhasoeva (Developer Relations @ ) -- developing . This article was written by Fabiana Clemente (Co-founder & CDO @ ) and Miriam Santos (Developer Relations @ ) -- developing -- and Zander Matheson (CEO & Founder @ YData YData ydata-profiling Bytewax Byetwax bytewax You may find additional information about the OSS packages in the respective documentations: & . ydata-profiling docs bytewax docs Also published here

How to Understand Your Data in Real-Time Using bytewax and ydata-profiling

Cómo entender sus datos en tiempo real usando bytewax y ydata-profiling

Comment comprendre vos données en temps réel en utilisant bytewax et ydata-profiling

बाइटवैक्स और वाईडेटा-प्रोफाइलिंग का उपयोग करके वास्तविक समय में अपने डेटा को कैसे समझें

bytewax と ydata-profiling を使用してデータをリアルタイムで理解する方法

Como entender seus dados em tempo real usando bytewax e ydata-profiling

Cách hiểu dữ liệu của bạn trong thời gian thực bằng cách sử dụng bytewax và ydata-profile

如何使用 bytewax 和 ydata-profiling 实时了解您的数据

The Confluent add-on for Ockam Orchestrator enables tamper-proof and end-to-end encrypted message streams through Confluent Cloud, with zero-code changes. The drop-in solution enables companies to exceed common security and compliance requirements without the need for costly rearchitecture or development work. Kafka deployments typically combine authentication tokens and Transport Layer Security (TLS) to protect data moving into and out of Kafka topics. While this does secure data in transit over the Internet, it doesn't provide a complete solution to securing data as it travels Kafka. The Kafka broker will be able to temporarily see the plaintext data. through Encrypting communication both into and out of your Kafka broker combined with encryption of data at rest, inside Kafka, won't be sufficient protection from a data breach if the Kafka broker or the infrastructure it is running on is compromised, as the plaintext data and the keys to decrypt the data are available in memory. Ockam solves these problems, while providing additional risk mitigating benefits and data integrity assurances, via the Confluent add-on for Ockam Orchestrator. Trust your data-in-motion Your team has decided that integrating a stream processing platform into your systems is the best way to meet both the scale and agility needs of your business, and Apache Kafka was the obvious choice for the platform to use. You've also decided to take advantage of using an experienced managed platform in Confluent Cloud so that your team can focus on adding value rather than running infrastructure. The development is complete, the team has a working solution where producers can send data to Confluent Cloud, and consumers are able to receive it on the other side. Communication between your producers, consumers, and Confluent Cloud is encrypted and secure using Transport Layer Security (TLS) so you're about to proceed with moving things into production usage. A security and compliance review has surfaced a problem. While trust Confluent Cloud, the increased focus on reducing supply-chain risk means that having data move through their systems unencrypted no longer meets your security requirements. How do you mitigate the risk to your business should some other vendor in your supply chain have a security breach? You need to be able to ensure that: you There is no possibility for a third-party to read the data while it's in transit A third-party can not modify the data while it's in transit Only authenticated and authorized parties can produce data Without solutions to both of these there is a risk that a security breach in your own supply-chain can not only expose your sensitive data, but cause unexpected downstream escalations of their own. Integrate with no code changes Let me take you through a complete working example of how, in just a few minutes, you can configure and integrate an entire end-to-end solution that ensures that your data is always encrypted and tamper-proof while in motion through Kafka and Confluent. Note: The code examples below may change as we continue to improve the developer experience. Please check for the most up to date documentation. https://docs.ockam.io/ Initial setup of Ockam If you’ve previously set up Ockam you can skip this section and move straight to the two examples below. brew install build-trust/ockam/ockam (if you’re not using brew for package management we have in our documentation) installation instructions for other systems Once installed you need to enroll your local identity with Ockam Orchestrator, run the command below and follow the instructions provided: ockam enroll Configure the Confluent add-on We're assuming here that you've already got a working cluster on Confluent Cloud (and the Kafka Command Line Tools that come with the ), so the first step is to configure your Ockam project to use the Confluent add-on by pointing it to your bootstrap server address: latest Kafka distribution ockam project addon configure confluent \
  --bootstrap-server YOUR_CONFLUENT_CLOUD_BOOTSTRAP_SERVER_ADDRESS We'll then need to save our Ockam project configuration so we can use it later to register our producers and consumer, so save the output to a file name : project.json ockam project information default --output json > project.json As the administrator of the Ockam project, you're able to control what other identities are allowed to enroll themselves into your project by issuing unique one-time use enrollment tokens. We'll start by creating one for our consumer:​ ockam project enroll --project-path project.json --attribute role=member > consumer.token ... and then one for the first producer: ockam project enroll --project-path project.json --attribute role=member > producer1.token The last configuration file we need to generate is , which will be where you store the username and password you use to access your cluster on Confluent Cloud: kafka.config cat > kafka.config <<EOF
request.timeout.ms=30000
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
        username="YOUR_CONFLUENT_CLOUD_USER_NAME" \
        password="YOUR_CONFLUENT_CLOUD_PASSWORD";
EOF Configure consumers to receive encrypted messages On your consumer node you'll start by creating a new identity (you'll need the Ockam Command installed, so repeat the install instructions if you're doing this on a separate host): ockam identity create consumer Copy the and files from the previous section, and then use them to authenticate and enroll this identity into your Ockam project: project.json consumer.token ockam project authenticate \
  --project-path project.json \
  --identity consumer \
  --token $(cat consumer.token) An Ockam node is a way to connect securely connect different services to each other, so we'll create one here that we'll use to communicate through the Confluent Cloud cluster using the identity we just created: ockam node create consumer \
  --project project.json 
  --identity consumer Once that completes we can now expose our Kafka bootstrap server. This is like the remote Kafka bootstrap server and brokers have become virtually adjacent on : localhost:4000 ockam service start kafka-consumer \
  --node consumer \
  --project-route /project/default \
  --bootstrap-server-ip 127.0.0.1 \
  --bootstrap-server-port 4000 \
  --brokers-port-range 4001-4100 Copy the file across, and use it to create a new topic that we'll use for sending messages between the producer and consumer in this demo (in this case we've called the topic ) kafka.config demo-topic kafka-topics.sh \
  --bootstrap-server localhost:4000 \
  --command-config kafka.config \
  --create \
  --topic demo-topic \
  --partitions 3 The final step is to start our consumer script, pointing it to as our bootstrap server: localhost:4000 kafka-console-consumer.sh \
  --topic demo-topic \
  --bootstrap-server localhost:4000 \
  --consumer.config kafka.config The consumer code will push all communication into the Ockam node process that is running on the local host. That local Ockam process will automatically manage the generation of cryptographic keys, establishing a secure channel for communication with any producer nodes, and then subsequently receiving, decrypting, and forwarding on any messages that are received from the broker running on our Confluent Cloud cluster. Send encrypted messages from multiple producers To have messages for our consumer to process, we need to have something producing them. We'll go through a very similar process now but instead create the parts necessary for a producer. We start once again by creating an identity on the producer's host (again, install the Ockam Command on that host if required): ockam identity create producer1 Copy over the and files from the earlier section and use it to authenticate and enroll into our Ockam project: project.json producer1.token ockam project authenticate \
  --project-path project.json \
  --identity producer1  \
  --token $(cat producer1.token) Create a node and link it to both the project and identity we've created: ockam node create producer1 \
  --project project.json \
  --identity producer1 And expose our Kafka bootstrap server on port so we can start sending messages through Confluent Cloud: 5000 ockam service start kafka-producer \
  --node producer1 \
  --project-route /project/default \
  --bootstrap-server-ip 127.0.0.1 \
  --bootstrap-server-port 5000 \
  --brokers-port-range 5001-5100 Make sure to copy the file across, and start your producer: kafka.config kafka-console-producer.sh \
  --topic demo-topic \
  --bootstrap-server localhost:5000 \
  --producer.config kafka.config Your existing producer code will now be running, communicating with the broker via the secure portal we've created that has exposed the Kafka bootstrap server and Kafka brokers on local ports, and sending messages through to the consumer that was setup in the previous step. However, all message payloads will be transparently encrypted as they enter the node on the producer, and not decrypted until they exit the consumer node. At no point in transit can the broker see the plaintext message payload that was initially sent by the producer. To take this example further you could repeat this step for number of producers by generating a new enrollment token for each, and then repeating the steps in this section. N If you look at the encrypted messages inside Confluent Cloud, they will render as unrecognizable characters like in the following screen capture: End-to-end encrypted message streams An example that only takes a few minutes to implement means it can be easy to miss a number of the important improvements to security and integrity that comes from this approach: : each consumer and producer generates its own cryptographic keys, and is issued its own unique credentials. They then use these to establish a mutually trusted secure channel between each other. By removing the dependency on a third-party service to store or distribute keys you're able to reduce your vulnerability surface area and eliminate single points of failure. Unique keys per identity : by pushing control of keys to the edges of the system, where authenticated encryption and decryption occurs, no other parties in the supply-chain are able to modify the data in transit. You can be assured that the data you receive at the consumer is exactly what was sent by your producers. You can also be assured that only authorized producers can write to a topic ensuring that the data in your topic is highly trustworthy. If you have even more stringent requirements you can take control of your credential authority and enforce granular authorization policies. Tamper-proof data transfer : Ockam secure channels regularly rotate authentication keys and session secrets. This approach means that if one of those session secrets was exposed your total data exposure window is limited to the small duration that secret was in use. Rotating authentication keys means that even when the identity keys of a producer are compromised - no historical data is compromised. You can selectively remove the compromised producer and its data. With centralized shared key distribution approaches there is the risk that all current and historical data can’t be trusted after a breach because it may have been tampered with or stolen. Ockam's approach eliminates the risk of compromised historical data and minimizes the risk to future data using automatically rotating keys. Reduced exposure window All these benefits are possible, today, with no code changes and minimal configuration change to existing deployments. Next steps If you haven't already followed the example in this guide, you can for free by following the instructions in our documentation. get started using Ockam If you’d like to talk specifically about this or other potential use cases for Ockam, the team would be more than happy to . chat with you Alternatively you can join us, and an ever growing community of developers who want to build trust by making applications that are secure-by-design, in the or on . Build Trust Discord server Github Also published . here

Stories, News & Mentions about DELTASTREAM

Deltastream’s Stories on HackerNoon