Skip to main content

习题 股票数据流处理

经过本章的学习,读者应该对Flink的DataStream API有了初步的认识,本节以股票价格这个场景来实践所学内容。

实验目的

针对具体的业务场景,学习如何定义相关数据结构,如何自定义Source,如何使用各类算子。

实验内容

我们虚构了一个股票交易数据集,如下所示,该数据集中有每笔股票的交易时间、价格和交易量。数据集放置在了src/main/resource/stock文件夹里。

股票代号,交易日期,交易时间(秒),价格,交易量
US2.AAPL,20200108,093003,297.260000000,100
US2.AAPL,20200108,093003,297.270000000,100
US2.AAPL,20200108,093003,297.310000000,100

数据类型和序列化章节,我们曾介绍Flink所支持的数据结构。对于这个股票价格的业务场景,首先要做的是对该业务进行建模,读者需要设计一个StockPrice类,能够表征一次交易数据。这个类至少包含以下字段:

/* 
* symbol 股票代号
* ts 时间戳
* price 价格
* volume 交易量
*/

接下来,我们自定义Source,这个自定义的类继承SourceFunction,读取数据集中的元素,并将数据写入DataStream<StockPrice>中。为了模拟不同交易之间的时间间隔,我们使用Thread.sleep()方法,等待一定的时间。下面的代码展示了如何自定义Source,读者可以直接拿来借鉴。

public class StockSource implements SourceFunction<StockPrice> {

// Source是否正在运行
private boolean isRunning = true;
// 数据集文件名
private String path;
private InputStream streamSource;

public StockSource(String path) {
this.path = path;
}

// 读取数据集中的元素,每隔时间发送一次股票数据
// 使用SourceContext.collect(T element)发送数据
@Override
public void run(SourceContext<StockPrice> sourceContext) throws Exception {
DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd HHmmss");
// 从项目的resources目录获取输入
streamSource = this.getClass().getClassLoader().getResourceAsStream(path);
BufferedReader br = new BufferedReader(new InputStreamReader(streamSource));
String line;
boolean isFirstLine = true;
long timeDiff = 0;
long lastEventTs = 0;
while (isRunning && (line = br.readLine()) != null) {
String[] itemStrArr = line.split(",");
LocalDateTime dateTime = LocalDateTime.parse(itemStrArr[1] + " " + itemStrArr[2], formatter);
long eventTs = Timestamp.valueOf(dateTime).getTime();
if (isFirstLine) {
// 从第一行数据提取时间戳
lastEventTs = eventTs;
isFirstLine = false;
}
StockPrice stock = StockPrice.of(itemStrArr[0], Double.parseDouble(itemStrArr[3]), eventTs, Integer.parseInt(itemStrArr[4]));
// 输入文件中的时间戳是从小到大排列的
// 新读入的行如果比上一行大,sleep,这样来模拟一个有时间间隔的输入流
timeDiff = eventTs - lastEventTs;
if (timeDiff > 0)
Thread.sleep(timeDiff);
sourceContext.collect(stock);
lastEventTs = eventTs;
}
}

// 停止发送数据
@Override
public void cancel() {
try {
streamSource.close();
} catch (Exception e) {
System.out.println(e.toString());
}
isRunning = false;
}
}

对于我们自定义的这个Source,我们可以使用下面的时间语义:

env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)

接下来,基于DataStream API,按照股票代号分组,对股票数据流进行分析和处理。

实验要求

完成数据结构定义和数据流处理部分的代码编写。其中数据流分析处理部分需要完成:

  • 程序1:价格最大值

实时计算某支股票的价格最大值。

  • 程序2:汇率转换

数据中股票价格以美元结算,假设美元和人民币的汇率为7,使用map进行汇率转换,折算成人民币。

  • 程序3:大额交易过滤

数据集中有交易量字段,给定某个阈值,过滤出交易量大于该阈值的,生成一个大额交易数据流。

实验报告

将思路和程序撰写成实验报告。