이미 스트리밍 된 Spark Streaming 앱 스트림 파일

Question

1 개의 이름 노드와 2 개의 데이터 노드가있는 YARN ec2 클러스터에 스파크 스트리밍 앱이 배포되었습니다. Google은 1 명의 핵심 집행자와 588MB의 RAM을 가진 11 명의 집행자와 함께 앱을 제출합니다. 지속적으로 쓰여지고있는 S3의 디렉토리에서 앱 스트리밍.

가

val ssc = new StreamingContext(sparkConf, Seconds(10)) 
val lines = ssc.fileStream[LongWritable, Text, TextInputFormat](Settings.S3RequestsHost , (f:Path)=> true, true) 
//some maps and other logic here 
ssc.start() 
ssc.awaitTermination() 

대신 textFileStream FILESTREAM의 사용 목적은 프로세스가 시작될 때 불꽃이 기존의 파일을 처리하는 방법을 정의한다 :이 달성한다는 코드의 라인이다. 프로세스를 시작한 후에 추가 된 새 파일 만 처리하고 기존 파일은 생략합니다. 우리는 배치 지속 시간을 10 초로 설정했습니다.

s3에 적은 수의 파일을 추가하는 동안 프로세스가 잘 진행됩니다. 4 또는 5라고 가정 해 봅시다. 스트리밍 UI에서 스테이지가 실행 파일에서 성공적으로 실행되는 방식을 볼 수 있습니다 (처리되는 각 파일에 대해 하나씩). 그러나 때로는 더 많은 수의 파일을 추가하려고 할 때 이상한 행동을합니다. 응용 프로그램은 이미 스트리밍 된 파일을 스트리밍하기 시작합니다.

예를 들어 s3에 20 개의 파일을 추가합니다. 파일은 3 일괄 처리됩니다. 첫 번째 일괄 처리는 7 개 파일, 두 번째 8 개 및 세 번째 5 개를 처리합니다.이 시점에서 S3에 추가 파일이 더 이상 존재하지 않지만 개이지만 더 이상 같은 파일을 사용하여이 단계를 반복적으로 시작합니다. 어떤 생각이 원인이 될 수 있습니까? https://issues.apache.org/jira/browse/SPARK-3553

Answer 1

주 문장 Spark Streaming Programming Guide에서 "파일은 원자 이동 또는 데이터 디렉토리로 이름을 변경하여 DataDirectory를 만들어야합니다"

나는이 문제에 대한 락스 티켓을 게시했습니다. 빈 파일을 만들고 추가하는 대신 전체 파일을 한 번에 모두 표시해야합니다.

한 가지 방법은 cloudberry에서 파일을 다른 곳에 저장 한 다음 정기적으로 스크립트를 실행하여 파일을 스트리밍 앱을 첨부 한 디렉토리로 이동하거나 이름을 변경하는 것입니다.