银行大数据业务怎么做数据的增量抽取,举例说明
时间: 2024-06-08 22:09:51 浏览: 14
银行大数据业务中,增量抽取是指只抽取最近一段时间内新增或修改的数据,而不是每次都全量抽取所有数据。这样做的好处是可以减少数据抽取的时间和资源消耗,同时也可以尽快反映出业务的变化。
举一个例子,比如一个银行的客户关系管理系统,该系统记录了银行与客户之间的各种交互信息,包括客户的个人信息、贷款、储蓄、信用卡等业务信息。该系统每天都会产生大量的数据,如何进行增量抽取呢?
首先,银行可以通过记录每个数据表的更新时间戳,比如每个表都有一个“last_update_time”的字段,记录上一次该表中数据更新的时间。然后,每次进行增量抽取时,只需要查询该表中“last_update_time”字段大于上一次抽取时间的数据即可,这样可以保证只抽取最近一段时间内新增或修改的数据。
另外,银行还可以通过增量抽取的方式,实现数据同步。比如,银行的核心业务系统每天会产生大量的交易数据,这些数据需要及时同步到数据仓库中,以供分析使用。使用增量抽取的方式,可以只抽取最近一段时间内新增或修改的交易数据,而不是每次都全量抽取,这样可以减少数据抽取的时间和资源消耗,同时也可以尽快反映出业务的变化。
相关问题
银行大数据平台做增量数据抽取,详细说明,带代码
银行的大数据平台需要从各个业务系统中抽取数据,以便进行数据分析和挖掘。增量数据抽取是一种常用的方式,可以定期抽取最新的数据,而不是全量抽取所有数据。以下是一种基于Apache Kafka和Apache Spark的增量数据抽取方案。
1. 准备工作
首先需要安装和配置以下工具和组件:
- Apache Kafka
- Apache Spark
- MySQL数据库
2. 创建Kafka主题
在Kafka中创建一个主题(topic)用于存储业务系统中的数据。可以使用Kafka命令行工具创建主题:
```
bin/kafka-topics.sh --create --topic bank-data --zookeeper localhost:2181 --replication-factor 1 --partitions 1
```
3. 编写数据生成器
为了模拟业务系统中的数据,需要编写一个数据生成器。可以使用Python编写一个简单的生成器,每隔一定时间生成一条随机的JSON格式数据,并将其发送到Kafka中。代码如下:
```python
import time
import json
from kafka import KafkaProducer
producer = KafkaProducer(bootstrap_servers='localhost:9092')
while True:
data = {
'account_number': random.randint(10000000, 99999999),
'transaction_amount': random.uniform(0, 10000),
'transaction_time': time.time()
}
producer.send('bank-data', json.dumps(data).encode())
time.sleep(1)
```
4. 创建MySQL表
为了存储增量数据抽取的状态信息,需要在MySQL中创建一个表。以下是一个简单的表结构:
```
CREATE TABLE incremental_extraction (
id INT PRIMARY KEY AUTO_INCREMENT,
last_offset BIGINT,
last_timestamp DATETIME
);
```
5. 编写Spark作业
使用Spark Streaming读取Kafka主题中的数据,并将其保存到MySQL中。为了实现增量数据抽取,需要记住上一次抽取的位置(即offset),并只读取新的数据。以下是一个基于Scala的Spark作业示例:
```scala
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.kafka.common.serialization.StringDeserializer
import java.sql.{Connection, DriverManager, ResultSet}
object IncrementalExtraction {
def main(args: Array[String]): Unit = {
val brokers = "localhost:9092"
val groupId = "incremental-extraction"
val topics = Set("bank-data")
val sparkConf = new SparkConf().setAppName("IncrementalExtraction")
val ssc = new StreamingContext(sparkConf, Seconds(1))
ssc.checkpoint("checkpoint")
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> brokers,
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> groupId,
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val jdbcUrl = "jdbc:mysql://localhost:3306/bank_data"
val jdbcUser = "root"
val jdbcPassword = "password"
var lastOffset = 0L
var lastTimestamp = new java.sql.Timestamp(0)
def getOffsetFromDb(): Unit = {
Class.forName("com.mysql.jdbc.Driver")
val conn = DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword)
val stmt = conn.createStatement()
val rs = stmt.executeQuery("SELECT * FROM incremental_extraction ORDER BY id DESC LIMIT 1")
if (rs.next()) {
lastOffset = rs.getLong("last_offset")
lastTimestamp = rs.getTimestamp("last_timestamp")
}
conn.close()
}
def saveOffsetToDb(offset: Long, timestamp: java.sql.Timestamp): Unit = {
Class.forName("com.mysql.jdbc.Driver")
val conn = DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword)
val stmt = conn.prepareStatement("INSERT INTO incremental_extraction (last_offset, last_timestamp) VALUES (?, ?)")
stmt.setLong(1, offset)
stmt.setTimestamp(2, timestamp)
stmt.executeUpdate()
conn.close()
}
getOffsetFromDb()
val messages = KafkaUtils.createDirectStream[String, String](
ssc,
LocationStrategies.PreferConsistent,
ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, Map("bank-data" -> lastOffset))
)
val transactions = messages.filter { message =>
val data = parseJson(message.value())
data("transaction_time") > lastTimestamp
}.map { message =>
parseJson(message.value())
}
transactions.foreachRDD { rdd =>
if (!rdd.isEmpty()) {
val lastMessage = rdd.reduce((a, b) => if (a("transaction_time") > b("transaction_time")) a else b)
lastOffset = lastMessage("offset").asInstanceOf[Long]
lastTimestamp = new java.sql.Timestamp(lastMessage("transaction_time").asInstanceOf[Double].toLong)
saveOffsetToDb(lastOffset, lastTimestamp)
rdd.foreachPartition { partitionOfRecords =>
Class.forName("com.mysql.jdbc.Driver")
val conn = DriverManager.getConnection(jdbcUrl, jdbcUser, jdbcPassword)
partitionOfRecords.foreach { record =>
val stmt = conn.prepareStatement("INSERT INTO transactions (account_number, transaction_amount, transaction_time) VALUES (?, ?, ?)")
stmt.setInt(1, record("account_number").asInstanceOf[Int])
stmt.setDouble(2, record("transaction_amount").asInstanceOf[Double])
stmt.setTimestamp(3, new java.sql.Timestamp(record("transaction_time").asInstanceOf[Double].toLong))
stmt.executeUpdate()
}
conn.close()
}
}
}
ssc.start()
ssc.awaitTermination()
}
def parseJson(jsonString: String): Map[String, Any] = {
JSON.parseFull(jsonString).get.asInstanceOf[Map[String, Any]]
}
}
```
该作业首先从MySQL中获取上一次抽取的位置和时间戳,然后使用Spark Streaming读取Kafka主题中的数据,并过滤出新的数据。最后将新的数据保存到MySQL中,并更新上一次抽取的位置和时间戳。
6. 运行作业
将以上代码保存为一个可运行的Jar包,并在命令行中运行以下命令启动Spark作业:
```
spark-submit --class IncrementalExtraction --master local[*] incremental_extraction.jar
```
该作业将会不断从Kafka中读取新的数据,并将其保存到MySQL中。通过修改数据生成器中的生成频率和数量,可以模拟不同的数据量和速度。同时,可以通过修改作业中的时间窗口大小和MySQL表结构来适应不同的场景和需求。
银行个贷业务日增量数据会有多大
银行个贷业务的日增量数据是受到很多因素的影响的,比如银行的规模、业务水平、市场需求等等。因此,不同银行之间的日增量数据是有很大差异的。一般而言,大型银行的个贷业务日增量可能会达到几千甚至几万笔,而小型银行可能会少很多。需要注意的是,由于银行个贷业务的复杂性和风险性,银行在进行个贷业务时需要进行充分的风险评估和审查,因此并非所有的申请都能够得到批准。