Squirrel 방식으로 Redis에 데이터 싱크
Apache Flink 와 Redis는 함께 사용하여 대량의 데이터를 처리할 수 있는 실시간 데이터 처리 파이프라인을 구축할 수 있는 두 가지 강력한 도구입니다. Flink는 데이터 스트림 처리를 위한 확장성이 뛰어난 내결함성 플랫폼을 제공하는 반면 Redis는 데이터를 저장하고 쿼리하는 데 사용할 수 있는 고성능 인메모리 데이터베이스를 제공합니다. 이 기사에서는 Flink를 사용하여 비동기 함수를 사용하여 Redis를 호출하는 방법을 살펴보고 비차단 방식으로 Redis에 데이터를 푸시하는 데 이를 사용할 수 있는 방법을 보여줍니다.
레디스 이야기
"Redis: 단순한 캐시 그 이상
Redis는 강력한 NoSQL 인메모리 데이터 구조 저장소로, 개발자가 자주 사용하는 도구입니다. 종종 단순한 캐시로 생각되지만 Redis는 그 이상입니다. 데이터베이스, 메시지 브로커 및 캐시 기능을 하나로 통합할 수 있습니다.
Redis의 강점 중 하나는 다양성입니다. 문자열, 목록, 집합, 정렬된 집합, 해시, 스트림, HyperLogLog 및 비트맵을 비롯한 다양한 데이터 유형을 지원합니다. Redis는 지리 공간 인덱스와 반경 쿼리도 제공하므로 위치 기반 애플리케이션에 유용한 도구입니다.
Redis의 기능은 데이터 모델을 넘어 확장됩니다. 기본 제공 복제, Lua 스크립팅 및 트랜잭션이 있으며 Redis 클러스터로 데이터를 자동으로 분할할 수 있습니다. 또한 Redis는 Redis Sentinel을 통해 고가용성을 제공합니다.
참고: 이 기사에서는 Redis 클러스터 모드에 더 중점을 둘 것입니다.
Redis 클러스터는 해시 슬롯과 함께 알고리즘 샤딩을 사용하여 주어진 키를 보유한 샤드를 결정하고 새 인스턴스 추가를 단순화합니다. 한편 Gossiping을 사용하여 클러스터의 상태를 확인하고 기본 노드가 응답하지 않는 경우 보조 노드를 승격하여 클러스터를 정상 상태로 유지할 수 있습니다. 스플릿 브레인 현상(클러스터가 승격할 사람을 결정할 수 없고 결국 스플릿 결정으로 끝나는 경우)을 방지하려면 강력한 설정을 위해 홀수 개의 기본 노드와 두 개의 복제본이 있어야 합니다.
Redis Cluster와 대화하기 위해 Redis Async Java 클라이언트인 lettuce를 사용할 것입니다 .
플링크 이야기
Apache Flink는 실시간, 높은 처리량 및 내결함성 데이터 처리를 처리하도록 설계된 오픈 소스, 통합 스트림 처리 및 일괄 처리 프레임워크입니다. Apache Gelly 프레임워크 위에 구축되었으며 제한된 스트림과 제한되지 않은 스트림에서 복잡한 이벤트 처리 및 상태 저장 계산을 지원하도록 설계되었습니다. 이를 빠르게 만드는 것은 메모리 내 성능을 활용하고 로컬 상태를 비동기적으로 체크포인트하는 것입니다.
이야기의 영웅
데이터베이스와의 비동기식 상호 작용은 스트림 처리 애플리케이션의 게임 체인저입니다. 이 접근 방식을 사용하면 단일 함수 인스턴스가 한 번에 여러 요청을 처리할 수 있으므로 동시 응답이 가능하고 처리량이 크게 향상됩니다. 대기 시간을 다른 요청 및 응답과 겹치게 함으로써 처리 파이프라인이 훨씬 더 효율적이 됩니다.
전자 상거래 데이터의 예를 사용하여 30초 슬라이드가 있는 24시간 슬라이딩 윈도우 에서 각 범주의 판매 수를 계산 하고 다운스트림 서비스에 대한 더 빠른 조회를 위해 Redis로 싱크합니다.
샘플 데이터 세트
Category, TimeStamp
Electronics,1679832334
Furniture,1679832336
Fashion,1679832378
Food,16798323536
package Aysnc_kafka_redis;
import AsyncIO.RedisSink;
import akka.japi.tuple.Tuple3;
import deserializer.Ecommdeserialize;
import model.Ecomm;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.util.concurrent.TimeUnit;
public class FlinkAsyncRedis {
public static void main(String[] args) throws Exception {
final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
Ecommdeserialize jsonde = new Ecommdeserialize();
KafkaSource<Ecomm> source = KafkaSource.<Ecomm>builder()
.setTopics("{dummytopic}")
.setBootstrapServers("{dummybootstrap}")
.setGroupId("test_flink")
.setStartingOffsets(OffsetsInitializer.earliest())
.setValueOnlyDeserializer(jsonde)
.build();
DataStream<Ecomm> orderData = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
orderData.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Ecomm>(Time.seconds(10)) {
@Override
public long extractTimestamp(Ecomm element) {
return element.getEventTimestamp(); // extract watermark column from stream
}
});
SingleOutputStreamOperator<Tuple3<String, Long, Long>> aggregatedData = orderData.keyBy(Ecomm::getCategory)
.window(SlidingEventTimeWindows.of(Time.hours(24),Time.seconds(30)))
.apply((WindowFunction<Ecomm, Tuple3<String, Long, Long>, String, TimeWindow>) (key, window, input, out) -> {
long count = 0;
for (Ecomm event : input) {
count++; // increment the count for each event in the window
}
out.collect(new Tuple3<>(key, window.getEnd(), count)); // output the category, window end time, and count
});
// calling async I/0 operator to sink data to redis in UnOrdered way
SingleOutputStreamOperator<String> sinkResults = AsyncDataStream.unorderedWait(aggregatedData,new RedisSink(
"{redisClusterUrl}"),
1000, // the timeout defines how long an asynchronous operation take before it is finally considered failed
TimeUnit.MILLISECONDS,
100); //capacity This parameter defines how many asynchronous requests may be in progress at the same time.
sinkResults.print(); // print out the redis set response stored in the future for every key
env.execute("RedisAsyncSink"); // you will be able to see your job running on cluster by this name
}
}
package AsyncIO;
import akka.japi.tuple.Tuple3;
import io.lettuce.core.RedisFuture;
import io.lettuce.core.cluster.RedisClusterClient;
import io.lettuce.core.cluster.api.StatefulRedisClusterConnection;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import lombok.AllArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import scala.collection.immutable.List;
import java.util.ArrayList;
import java.util.Collections;
@AllArgsConstructor
public class RedisSink extends RichAsyncFunction<Tuple3<String, Long, Long>, String> {
String redisUrl;
public RedisSink(String redisUrl){
this.redisUrl=redisUrl;
}
private transient RedisClusterClient client = null;
private transient StatefulRedisClusterConnection<String, String> clusterConnection = null;
private transient RedisAdvancedClusterAsyncCommands<String, String> asyncCall = null;
// method executes any operator-specific initialization
@Override
public void open(Configuration parameters) {
if (client == null ) {
client = RedisClusterClient.create(redisUrl);
}
if (clusterConnection == null) {
clusterConnection = client.connect();
}
if (asyncCall == null) {
asyncCall = clusterConnection.async();
}
}
// core logic to set key in redis using async connection and return result of the call via ResultFuture
@Override
public void asyncInvoke(Tuple3<String, Long, Long> stream, ResultFuture<String> resultFuture) {
String productKey = stream.t1();
System.out.println("RedisKey:" + productKey); //for logging
String count = stream.t3().toString();
System.out.println("Redisvalue:" + count); //for logging
RedisFuture<String> setResult = asyncCall.set(productKey,count);
setResult.whenComplete((result, throwable) -> {if(throwable!=null){
System.out.println("Callback from redis failed:" + throwable);
resultFuture.complete(new ArrayList<>());
}
else{
resultFuture.complete(new ArrayList(Collections.singleton(result)));
}});
}
// method closes what was opened during initialization to free any resources
// held by the operator (e.g. open network connections, io streams)
@Override
public void close() throws Exception {
client.close();
}
}
- Redis로 스트리밍된 데이터는 데이터 과학 모델에서 세일 시즌 동안 자주 판매되는 카테고리에 대해 더 많은 제품을 조회하고 생성하는 데 사용할 수 있습니다.
- 웹 페이지에서 차트와 숫자를 판매 통계로 표시하여 공격적인 구매를 유도하는 데 사용할 수 있습니다.
- Flink는 데이터 스트림 처리를 위한 확장성이 뛰어난 내결함성 플랫폼을 제공하는 반면 Redis는 데이터를 저장하고 쿼리하는 데 사용할 수 있는 고성능 인메모리 데이터베이스를 제공합니다.
- 비동기 프로그래밍은 Redis와 같은 외부 시스템에 대한 비차단 호출을 허용하여 데이터 처리 파이프라인의 성능을 개선하는 데 사용할 수 있습니다.
- 이 두 가지를 결합하면 실시간 데이터 결정 문화를 가져오는 데 도움이 될 수 있습니다.
https://architecturenotes.co/redis/.
https://www.baeldung.com/java-redis-lettuce
https://nightlies.apache.org/flink/flink-docs-release-1.16/docs/dev/datastream/operators/asyncio/
![연결된 목록이란 무엇입니까? [1 부]](https://post.nghiatu.com/assets/images/m/max/724/1*Xokk6XOjWyIGCBujkJsCzQ.jpeg)
