Table-Computing 开源项目教程

2024-08-07 00:05:50作者：仰钰奇

Table-Computing (Simplified as TC) is a high performance and low latency computing framework, 10x faster than Flink for complicated use cases, distributed and light weighted, relational operation, simple to use, write less and do more.

项目地址：https://gitcode.com/gh_mirrors/ta/table-computing

项目介绍

Table-Computing（简称 TC）是一个高性能、低延迟的计算框架，专为复杂用例设计，比 Flink 快 10 倍。它是一个分布式、轻量级的关系操作框架，简单易用，能够实现“写得更少，做得更多”。TC 主要用于大数据流处理和数据分析，支持复杂的计算任务，如滑动窗口计算、数据重哈希和重平衡等。

项目快速启动

环境准备

Java 8 或更高版本
Maven
Kafka（用于流数据输入）
MySQL（用于维度表数据）

快速启动步骤

克隆项目

git clone https://github.com/alibaba/table-computing.git
cd table-computing

构建项目
```
mvn clean install
```

配置和运行示例

以下是一个简单的示例，展示如何使用 TC 进行滑动窗口计算。

import com.alibaba.table_computing.*;

public class QuickStart {
    public static void main(String[] args) {
        // 配置 Kafka 流表
        KafkaStreamTable kafkaStreamTable = new KafkaStreamTable(
            "localhost:9092", "consumerGroupId", "topic", 0, columnTypeMap);
        kafkaStreamTable.start();

        // 配置 MySQL 维度表
        MysqlDimensionTable mysqlDimensionTable = new MysqlDimensionTable(
            "jdbc:mysql://localhost:3306/e-commerce", "commodity", "userName", "password",
            Duration.ofHours(1), new ColumnTypeBuilder()
                .column("id", Type.INT)
                .column("name", Type.VARCHAR)
                .column("price", Type.INT)
                .build(), "id");

        // 创建流处理实例
        StreamProcessing sp = new StreamProcessing();

        // 配置滑动窗口
        String[] hashBy = new String[] {"commodity_id"};
        SlideWindow slideWindow = new SlideWindow(
            Duration.ofHours(1), Duration.ofMinutes(30), hashBy, "__time__",
            new AggTimeWindowFunction() {
                @Override
                public Comparable[] agg(List<Comparable> partitionByColumns, List<Row> rows, long windowStart, long windowEnd) {
                    return new Comparable[] {
                        partitionByColumns.get(0),
                        AggregationUtil.sumInt(rows, "count"),
                        AggregationUtil.sumInt(rows, "total_price"),
                        windowStart
                    };
                }
            }, new String[] {"commodity_id", "sales_volume", "saleroom", "window_start"});

        slideWindow.setWatermark(Duration.ofSeconds(2));

        // 启动流处理
        sp.start();
    }
}

应用案例和最佳实践

案例一：电商销售数据分析

在电商平台上，使用 TC 可以实时计算每小时的销售量和销售额，生成销售排行榜。以下是一个示例代码：

// 配置 Kafka 流表和 MySQL 维度表（如上所示）

// 创建流处理实例
StreamProcessing sp = new StreamProcessing();

// 配置滑动窗口
String[] hashBy = new String[] {"commodity_id"};
SlideWindow slideWindow = new SlideWindow(
    Duration.ofHours(1), Duration.ofMinutes(30), hashBy, "__time__",
    new AggTimeWindowFunction() {
        @Override
        public Comparable[] agg(List<Comparable> partitionByColumns, List<Row> rows, long windowStart, long windowEnd) {
            return new Comparable[] {
                partitionByColumns.get(0),
                AggregationUtil.sumInt(rows, "count"),
                AggregationUtil.sumInt(rows, "total_price"),
                windowStart
            };
        }
    }, new String[] {"commodity_id", "sales_volume", "saleroom", "window_start"});

slideWindow.setWatermark(Duration.ofSeconds(2));

// 启动流处理
sp.start();