fire

README

<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements.  See the NOTICE file
distributed with this work for additional information
regarding copyright ownership.  The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License.  You may obtain a copy of the License at

  http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied.  See the License for the
specific language governing permissions and limitations
under the License.
-->

<p align="center">
  <a href="https://www.fireframework.cn/">
    <img src="https://web-code-img.oss-cn-shanghai.aliyuncs.com/readme/49c2cb65ea2a38f5b5cc00ce.jpg" width="600" height="600">
  </a>
  <p align="center">
    <a href="https://github.com/FireFramework/fire/stargazers">
    <img src="https://img.shields.io/github/stars/FireFramework/fire?style=flat-square&logo=github" alt="GitHub stars">
    </a>
    <a href="https://github.com/FireFramework/fire/network/members">
     <img src="https://img.shields.io/github/forks/FireFramework/fire?style=flat-square&logo=github" alt="GitHub forks">
    </a>
    <a href='https://gitee.com/fire-framework/fire/stargazers'><img src='https://gitee.com/fire-framework/fire/badge/star.svg?theme=gvp' alt='Gitee star'></img></a>
    <a href='https://gitee.com/fire-framework/fire/members'><img src='https://gitee.com/fire-framework/fire/badge/fork.svg?theme=gvp' alt='Gitee fork'></img></a>
		<a href='https://gitcode.com/fire-framework/fire'><img src='https://gitcode.com/fire-framework/fire/star/badge.svg' alt='GitCode star'></img></a>
  </p>
</p>


# Fire框架

　　Fire框架是由**中通**大数据自主研发并开源的、专门用于进行Spark和Flink任务开发的大数据框架，可节约70%以上的代码量。首创基于注解进行Spark和Flink任务开发，具备**实时血缘、根因诊断、动态调优、参数热调整**等众多平台化功能。Fire框架在中通内部每天处理数据量高达数千亿，在外部已被数十家公司所使用。

## 一、就这么简单！

### 1.1 Flink开发示例

```scala
@Config(
  """
    |state.checkpoints.num-retained=30 	# 支持任意Flink调优参数、Fire框架参数、用户自定义参数等
    |state.checkpoints.dir=hdfs:///user/flink/checkpoint
    |""")
@Hive("thrift://localhost:9083") // 配置连接到指定的hive
@Streaming(interval = 100, unaligned = true) // 100s做一次checkpoint，开启非对齐checkpoint
@Kafka(brokers = "localhost:9092", topics = "fire", groupId = "fire")
object FlinkDemo extends FlinkStreaming {

  @Process
  def kafkaSource: Unit = {
    val dstream = this.fire.createKafkaDirectStream() 	// 使用api的方式消费kafka
    sql("""create table statement ...""")
    sql("""insert into statement ...""")
  }
}
```

### 1.2 Spark开发示例

```scala
@Config(
  """
    |spark.shuffle.compress=true		# 支持任意Spark调优参数、Fire框架参数、用户自定义参数等
    |spark.ui.enabled=true
    |""")
@Hive("thrift://localhost:9083") // 配置连接到指定的hive
@Streaming(interval = 100, maxRatePerPartition = 100) // 100s一个Streaming batch，并限制消费速率
@Kafka(brokers = "localhost:9092", topics = "fire", groupId = "fire")
object SparkDemo extends SparkStreaming {

  @Process
  def kafkaSource: Unit = {
    val dstream = this.fire.createKafkaDirectStream() 	// 使用api的方式消费kafka
    sql("""select * from xxx""").show()
  }
}
```

***说明：structured streaming、spark core、flink sql、flink批任务均支持，代码结构与上述示例一致。***

## *二、开发与示例*

## 2.1 [Spark开发示例](https://gitee.com/fire-framework/spark-examples)

## 2.2 [Flink开发示例](https://gitee.com/fire-framework/flink-examples)

**示例项目clone后导入idea即可run，无需任何额外配置！**

## 三、亮点多多！

### 3.1 兼容主流版本

　　fire框架适配了不同的spark与flink版本，支持spark2.x及以上所有版本，flink1.10及以上所有版本，支持基于scala2.11或scala2.12进行编译。

```shell
# 可根据实际需要选择不同的引擎版本进行fire框架的构建
mvn clean install -DskipTests -Pspark-3.0 -Pflink-1.14 -Pscala-2.12
```

| Apache Spark | Apache Flink |
| ------------ | ------------ |
| 2.3.x        | 1.12.x       |
| 2.4.x        | 1.13.x       |
| 3.0.x        | 1.14.x       |
| 3.1.x        | 1.15.x       |
| 3.2.x        | 1.16.x       |
| 3.3.x        | 1.17.x       |
| 3.4.x        | 1.18.x       |
| 3.5.x        | 1.19.x       |

### **3.2 简单好用**

　　Fire框架高度封装，屏蔽大量技术细节，许多connector仅需一行代码即可完成主要功能。同时Fire框架统一了spark与flink两大引擎常用的api，使用统一的代码风格即可实现spark与flink的代码开发。

- **HBase API**

```scala
// 读取HBase中指定rowkey数据并将结果集封装为DataFrame返回
val studentDF: DataFrame = this.fire.hbaseGetDF(hTableName, classOf[Student], getRDD)
// 将指定数据集分布式插入到指定HBase表中
this.fire.hbasePutDF(hTableName, studentDF, classOf[Student])
```

- **JDBC API**

1. **通过注解配置数据源：**

```java
@Jdbc(url = "jdbc:mysql://mysql-server:3306/fire", username = "root", password = "root")
```

2. **Spark示例：**

```scala
// 将DataFrame中指定几列插入到关系型数据库中，每100条一插入
df.jdbcBatchUpdate(insertSql, Seq("name", "age", "createTime", "length", "sex"), batch = 100)
// 将查询结果通过反射映射到DataFrame中
val df: DataFrame = this.fire.jdbcQueryDF(querySql, Seq(1, 2, 3), classOf[Student])

```

3. **Flink示例：**

```scala
val dstream = this.fire.createKafkaDirectStream().map(t => JSONUtils.parseObject[Student](https://gitee.com/fire-framework/fire/blob/master/t))
val sql =
s"""
|insert into spark_test(name, age, createTime) values(?, ?, ?)
|ON DUPLICATE KEY UPDATE age=18
|""".stripMargin
// sinkJdbc只需指定sql语句即可，fire会自动推断sql中占位符与JavaBean中成员变量的对应关系
dstream.sinkJdbc(sql)
dstream.sinkJdbcExactlyOnce(sql, keyNum = 2)
```

### **3.3 灵活的配置方式**

　　支持基于接口、apollo、配置文件以及注解等多种方式配置，支持将spark&flink等**引擎参数**、**fire框架参数**以及**用户自定义参数**混合配置，支持运行时动态修改配置。几种常用配置方式如下（[配置手册](https://gitee.com/fire-framework/fire/blob/master/docs/config.md)）：

1. **基于配置文件：** 创建类名同名的properties文件进行参数配置
2. **基于接口配置：** fire框架提供了配置接口调用，通过接口获取所需的配置，可用于平台化的配置管理
3. **基于注解配置：** 通过注解的方式实现集群环境、connector、调优参数的配置，常用注解如下：

```scala
@Config(
  """
    |# 支持Flink调优参数、Fire框架参数、用户自定义参数等
    |state.checkpoints.num-retained=30
    |state.checkpoints.dir=hdfs:///user/flink/checkpoint
    |""")
@Hive("thrift://localhost:9083")
@Checkpoint(interval = 100, unaligned = true)
@Kafka(brokers = "localhost:9092", topics = "fire", groupId = "fire")
@RocketMQ(brokers = "bigdata_test", topic