File Source 커넥터
File Source 커넥터
File source 커넥터는 디렉터리에 있는 파일에서 메시지를 읽어 Pulsar 토픽에 저장하는 커넥터예요. 파일 기반의 데이터를 Pulsar로 가져올 때 사용해요.
참고: 모든 Pulsar 커넥터는 download page에서 내려받을 수 있어요.
출처: 문서
본문
구성 (Configuration)
File source 커넥터의 구성에는 다음과 같은 프로퍼티가 있어요.
프로퍼티 (Property)
| 이름 | 타입 | 필수 | 기본값 | 설명 |
|---|---|---|---|---|
| inputDirectory | String | true | 기본값 없음 | 파일을 가져올 입력 디렉터리예요. |
| recurse | Boolean | false | true | 하위 디렉터리에서도 파일을 가져올지 여부예요. |
| keepFile | Boolean | false | false | true로 설정하면 파일을 처리한 뒤에도 삭제하지 않아요. 즉 파일을 계속해서 다시 처리할 수 있다는 뜻이에요. |
| fileFilter | String | false | [^\].* | 주어진 정규식과 이름이 일치하는 파일을 처리해요. |
| pathFilter | String | false | NULL | recurse가 true로 설정되면, 경로가 주어진 정규식과 일치하는 하위 디렉터리를 스캔해요. |
| minimumFileAge | Integer | false | 0 | 파일을 처리할 수 있는 최소 나이(age)예요. minimumFileAge보다 어린 파일(마지막 수정 날짜 기준)은 무시돼요. |
| maximumFileAge | Long | false | Long.MAX_VALUE | 파일을 처리할 수 있는 최대 나이예요. maximumFileAge보다 오래된 파일(마지막 수정 날짜 기준)은 무시돼요. |
| minimumSize | Integer | false | 1 | 파일을 처리할 수 있는 최소 크기(바이트)예요. |
| maximumSize | Double | false | Double.MAX_VALUE | 파일을 처리할 수 있는 최대 크기(바이트)예요. |
| ignoreHiddenFiles | Boolean | false | true | 숨김 파일을 무시할지 여부예요. |
| pollingInterval | Long | false | 10000L | 디렉터리 목록을 조회하기 전에 기다리는 시간이에요. |
| numWorkers | Integer | false | 1 | 파일을 처리하는 워커 스레드 수예요. 이 값으로 여러 파일을 동시에 더 많이 처리할 수 있어요. 다만 1보다 큰 값으로 설정하면 여러 파일의 데이터가 대상 토픽에 섞여 들어가요. |
| processedFileSuffix | String | false | NULL | 설정하면 처리된 파일을 삭제하지 않고 이름만 변경해요. 이 구성은 'keepFile' 프로퍼티가 false일 때만 동작해요. |
예제 (Example)
File source 커넥터를 사용하기 전에 다음 방법 중 하나로 구성 파일을 만들어야 해요.
- JSON
{
"configs": {
"inputDirectory": "/Users/david",
"recurse": true,
"keepFile": false,
"fileFilter": "[^\\.].*",
"pathFilter": ".*",
"minimumFileAge": 0,
"maximumFileAge": 9999999999,
"minimumSize": 1,
"maximumSize": 5000000,
"ignoreHiddenFiles": true,
"pollingInterval": 5000,
"numWorkers": 1,
"processedFileSuffix": ".processed_done"
}
}
- YAML
configs:
inputDirectory: "/Users/david"
recurse: true
keepFile: false
fileFilter: "[^\\.].*"
pathFilter: ".*"
minimumFileAge: 0
maximumFileAge: 9999999999
minimumSize: 1
maximumSize: 5000000
ignoreHiddenFiles: true
pollingInterval: 5000
numWorkers: 1
processedFileSuffix: ".processed_done"
사용법 (Usage)
여기 File source 커넥터를 사용하는 예시가 있어요.
- Pulsar 이미지를 내려받아요.
docker pull apachepulsar/pulsar:{version}
- Pulsar standalone을 시작해요.
docker run -d -it -p 6650:6650 -p 8080:8080 -v $PWD/data:/pulsar/data --name pulsar-standalone apachepulsar/pulsar:{version} bin/pulsar standalone
- file-connector.yaml 구성 파일을 만들어요.
configs:
inputDirectory: "/opt"
- 구성 파일 file-connector.yaml을 컨테이너로 복사해요.
docker cp connectors/file-connector.yaml pulsar-standalone:/pulsar/
- File source 커넥터를 다운로드해요.
curl -O https://mirrors.tuna.tsinghua.edu.cn/apache/pulsar/pulsar-{version}/connectors/pulsar-io-file-{version}.nar
connectors폴더로 복사한 뒤 컨테이너를 재시작해요.
docker cp pulsar-io-file-{version}.nar pulsar-standalone:/pulsar/connectors/
docker restart pulsar-standalone
- File source 커넥터를 시작해요.
docker exec -it pulsar-standalone /bin/bash
./bin/pulsar-admin sources localrun \
--archive /pulsar/connectors/pulsar-io-file-{version}.nar \
--name file-test \
--destination-topic-name pulsar-file-test \
--source-config-file /pulsar/file-connector.yaml
- 컨슈머를 시작해요.
./bin/pulsar-client consume -s file-test -n 0 pulsar-file-test
- test.txt 파일에 메시지를 써요.
echo "hello world!" > /opt/test.txt
컨슈머 터미널 창에 다음 정보가 나타나요.
----- got message -----
hello world!