Kinesis S3へデータを書き込むサンプルアプリケーション

参考

Kinesisってなんじゃ？（Java実装編）

サンプルアプリケーション実装

前回につづいて、S3へファイルを書き込むアプリケーションの実装サンプル。
同じように、AwsCredentials.properties又は、EC2ロールで、S3へのアクセス権限を持つものに設定をしておくこと。サンプルではとりあえず、PowerUserのロールで行っています。

SampleS3putter.java

	private static final Logger LOG = LoggerFactory
			.getLogger(SampleS3Putter.class);

	public SampleS3Putter() {
	}

	private AmazonS3Client s3;
	private static final String BUCKET_NAME = "<バケット名>";
	private static final String DIRECTORY_PATH = "log/";
	private static final String EXTENSION = ".txt";
	private final CharsetDecoder decoder = Charset.forName("UTF-8")
			.newDecoder();
	private static final long CHECKPOINT_INTERVAL_MILLIS = 60000L;
	private long nextCheckpointTimeInMillis;

	@Override
	public void initialize(String shardId) {
        try {
            credentialsProvider = new InstanceProfileCredentialsProvider();
            credentialsProvider.getCredentials();
        } catch (AmazonClientException e) {
            credentialsProvider = new ClasspathPropertiesFileCredentialsProvider();
            credentialsProvider.getCredentials();
        }
		this.s3 = new AmazonS3Client(credentialsProvider);
	}

	@Override
	public void processRecords(List<Record> records,
			IRecordProcessorCheckpointer checkpointer) {
		process(records);
		if (System.currentTimeMillis() > nextCheckpointTimeInMillis) {
		　　　　　try {
			　　　checkpointer.checkpoint();
	　　　　　	} catch (ShutdownException se) {
			　　// 処理
	　　　　　	} catch (ThrottlingException e) {
			　　// 処理
	　　　　　	} catch (InvalidStateException e) {
			　　// 処理
	　　　　　	}
			nextCheckpointTimeInMillis = System.currentTimeMillis()
					+ CHECKPOINT_INTERVAL_MILLIS;
		}
	}

	private void process(List<Record> records) {

		if (records.isEmpty()) {
			LOG.info("record empty.");
			return;
		}

		byte[] dst = new byte[0];
		for (Record record : records) {
			LOG.debug(
					"---- Data : {} ,PartitionKey : {} , SequenceNumber : {} ----",
					record.getData(), record.getPartitionKey(),
					record.getSequenceNumber());

			try {
				String data = decoder.decode(record.getData()).toString()
						+ "\n";
				byte[] dstBytes = dst;
				byte[] dataBytes = data.getBytes("UTF-8");
				dst = new byte[dstBytes.length + dataBytes.length];
				System.arraycopy(dstBytes, 0, dst, 0, dstBytes.length);
				System.arraycopy(dataBytes, 0, dst, dstBytes.length,
						dataBytes.length);
			} catch (CharacterCodingException e) {
				// 何か処理
			} catch (UnsupportedEncodingException e) {
				// 何か処理
			}

		}

		// s3への書き込み
		ObjectMetadata metadata = new ObjectMetadata();
		metadata.setContentLength(dst.length);
		PutObjectResult putObjectResult = s3.putObject(BUCKET_NAME,
				DIRECTORY_PATH + System.currentTimeMillis()+"_log"+ EXTENSION,
				new ByteArrayInputStream(dst), metadata);

		LOG.debug(putObjectResult.toString());
	}

	@Override
	public void shutdown(IRecordProcessorCheckpointer checkpointer,
			ShutdownReason reason) {
		if (reason == ShutdownReason.TERMINATE) {
		　　　　　try {
			　　　checkpointer.checkpoint();
	　　　　　	} catch (ShutdownException se) {
			　　// 処理
	　　　　　	} catch (ThrottlingException e) {
			　　// 処理
	　　　　　	} catch (InvalidStateException e) {
			　　// 処理
	　　　　　	}
		}
	}

Factoryクラス

S3PutterRecordProcessorFactory.java

	public S3PutterRecordProcessorFactory() {
	}

	/* (非 Javadoc)
	 * @see com.amazonaws.services.kinesis.clientlibrary.interfaces.IRecordProcessorFactory#createProcessor()
	 */
	@Override
	public IRecordProcessor createProcessor() {
		return new SampleS3Putter();
	}

前回と同じように起動するようにすれば、いけるはず。

実装のポイント

1. AmazonS3Clientを使う。S3にアクセスできる権限をもったユーザでアクセスする。
2. 一度に出力するデータ（レコード数）を考える。100レコードで一回出力するとか。データのサイズで出力するなど。

また何かあったら追記

次く。。。