나는 Kafka 스트리밍을 처리하기 위해 flink 작업을 점심으로 먹기위한 Java 응용 프로그램을 가지고 있습니다. 작업이 카프카에서 들어오는 streamings 영원히 실행되기 때문에 응용 프로그램는 flinkEnv.execute("flink job name") 여기 작업 제출에서 대기 중입니다. 이 경우 실행에서 작업 ID가 어떻게 반환 될까요? j
를 시작 얻을 : mappedUserTrackingEvent
.keyBy("videoId", "userId")
.timeWindow(Time.seconds(30))
.sum("count")
실제로 너무 키 필드로 윈도우 시작 시간을 추가하고 싶습니다. 결과는 다음과 같을 것입니다 : key: videoId=123,u
Flink CEP 라이브러리를 사용하여 Hello 및 world가 발견되면 문자열을 인쇄하려고합니다. 내 소스는 카프카 (Kafka)이며 콘솔 생산자를 사용하여 데이터를 입력합니다. 그 부분이 효과가 있습니다. 내가 주제에 입력 한 것을 인쇄 할 수 있습니다. 그러나, 그것은 나의 최종 메시지를 인쇄하지 않을 것이다 "세계는 매우 멋지다!". 그것이 람다에
데이터 스트리밍을 위해 apache flink에서 작업 중이며 몇 가지 질문이 있습니다. 어떤 도움이라도 대단히 감사합니다. 감사. 1) 텀블링 창을 만드는 데 제한이 있습니까? 예를 들어, 사용자 ID 당 2 초간 텀블링 창을 만들고 싶다면 1 천만 개 이상의 사용자 ID가 문제가된다고 가정 해 봅시다. (나는 keyBy 사용자 ID를 사용하고 2 초 동
나중에 작업을 막을 수 있도록 작업에 고유 한 이름을 붙일 수 있습니까? grep을하고 작업 ID를 유지하고 싶지 않습니다. 간단히 말해서 배치의 일부로 작업을 중지하고 새 배치를 원합니다. 예를 들어, 당신이 execute(name: String) 호출을 시작할 때이 작업의 이름을 지정할 수 있습니다
기존 카탈로그 프로세서를 다시 설계해야하며 요구 사항은 다음과 같습니다. 요구 사항 저장소마다 'XML'파일을 제공하는 5 ~ 10 개의 공급 업체 (각 공급 업체는 여러 매장을 보유 할 수 있음)가 있습니다. 기본적으로 Store 당 1 개의 제품 xml 파일과 공급 업체 당 여러 개의 Store 파일. 최대 파일 크기는 500MB, 최소값은 100MB
필자의 요구 사항은 하루에 수백만 개의 레코드를 스트리밍하는 것이며 외부 구성 매개 변수에 큰 의존성이 있습니다. 예를 들어, 사용자는 웹 응용 프로그램에서 언제든지 필요한 설정을 변경할 수 있으며 변경 후에는 새로운 응용 프로그램 구성 매개 변수로 스트리밍해야합니다. 이는 앱 수준의 구성이며 각 데이터를 전달하고 필터링해야하는 일부 동적 제외 매개 변수도
이 게이지를 사용할 수 없습니다 추가 할 때 빌드가 실패, 빌드가 실패 : def open(configuration: Configuration) {
getRuntimeContext()
.getMetricGroup()
.gauge("RecordConverter.latency", new Gauge[Int]() {
@Over
병렬 처리가 8 인 MapStream이 있습니다. 두 개의 싱크를 DataStream에 추가합니다. 하나는 느리고 (Elasticsearch) 다른 하나는 빠릅니다 (HDFS). 그러나 내 이벤트는 ES에 플러시 된 후에 만 HDFS에 기록되므로 ES가없는 경우보다 ES의 경우 시간이 오래 걸립니다. dataStream.setParallelism(8);
파티션이 3 인 3 개의 주제와 연결을 시도한 경우 FlinkKafkaConsumer09은 하나의 주제에서 소비하고 아래의 Kafka 소비자 그룹 속성을 사용합니다. props.setProperty("group.id", "myGroup");
props.setProperty("auto.offset.reset", "latest");
그러나 여전히