تسلط بر Apache Flink: ستون فقرات پایپلاینهای داده در زمان واقعی
در منظره دادههای مدرن، پردازش دستهای دیگر کافی نیست. سازمانها به بینشهای فوری نیاز دارند و سیستمهایی را میطلبند که رویدادها را همانطور که رخ میدهند پردازش کنند. Apache Flink به عنوان استاندارد غیررسمی برای این نیاز ظهور کرده است و یک موتور قدرتمند و با عملکرد بالا برای محاسبات حالتدار روی جریانهای داده محدود و نامحدود ارائه میدهد. این پست به بررسی مکانیزمهای اصلی میپردازد که Flink را به ابزاری قدرتمند برای توسعهدهندگان سطح متوسط تا پیشرفته تبدیل میکند.
stream = inputStream.assignTimestampsAndWatermarks(
WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofSeconds(5))
.withTimestampAssigner((event, timestamp) -> event.getTimestamp())
);
```
در مثال بالا، ما تا ۵ ثانیه تأخیر در ترتیب رویدادها را تحمل میکنیم. این اطمینان حاصل میکند که پنجرهها تنها زمانی بسته میشوند که سیستم مطمئن باشد رویدادهای دیررس قابل توجهی در راه نیستند، که این امر امکان جمعآوری دقیق را بدون قربانی کردن بیش از حد تأخیر فراهم میکند.
counts = input
.keyBy(value -> value.getId())
.flatMap(new CountWindowAverage());
public static class CountWindowAverage implements FlatMapFunction, Tuple2> {
private transient ValueState sum;
@Override
public void open(Configuration parameters) {
ValueStateDescriptor descriptor = new ValueStateDescriptor<>("average-sum", Long.class);
sum = getRuntimeContext().getState(descriptor);
}
@Override
public void flatMap(Tuple2 value, Collector> out) throws Exception {
Long currentSum = sum.value();
currentSum += value.f1;
sum.update(currentSum);
// Emit average every 5 elements
if (/* count logic */) {
out.collect(Tuple2.of(value.f0, currentSum / 5));
}
}
}
```
این قطعه کد نشان میدهد که چگونه میتوان یک مجموع جاری را برای هر کلید حفظ کرد. حالت توسط بکاند حالت Flink (RocksDB یا HashMap) پشتیبانی میشود و در برابر خطاها مقاوم است.
pattern = Pattern.begin("start")
.where(new SimpleCondition() {
@Override
public boolean filter(Event value) { return value.name().equals("login"); }
})
.next("middle")
.where(new SimpleCondition() {
@Override
public boolean filter(Event value) { return value.amount() > 1000; }
})
.within(Time.seconds(10));
```
این الگو به دنبال یک رویداد "ورود" دنبالشده توسط یک تراکنش با ارزش بالا در عرض ۱۰ ثانیه میگردد که این امر تشخیص تقلب در زمان واقعی را امکانپذیر میسازد.