有什么方法可以防止 camus 增加偏移值，使其始终读取上一个值？以及如何为创建的文件提供一个自定义名称，而不是通用格式。 topic.leaderId.partitionId.partitioner.encodePartition(context, key) ?
camus
topic.leaderId.partitionId.partitioner.encodePartition(context, key)
rmbxnbpk1#
在中生成Map器输出的文件名 generateFileName 方法 com.linkedin.camus.etl.Patitioner 班级。您可以实现一个自定义分区器，并在中提供一个逻辑 @Override public String generateFileName(JobContext context, String topic, String brokerId, int partitionId, int count, long offset, String encodedPartition) { return "MyCustomFileName-"+topic+partitionId; }
generateFileName
com.linkedin.camus.etl.Patitioner
@Override public String generateFileName(JobContext context, String topic, String brokerId, int partitionId, int count, long offset, String encodedPartition) { return "MyCustomFileName-"+topic+partitionId; }
1条答案
按热度按时间rmbxnbpk1#
在中生成Map器输出的文件名
generateFileName方法com.linkedin.camus.etl.Patitioner班级。您可以实现一个自定义分区器，并在中提供一个逻辑
@Override public String generateFileName(JobContext context, String topic, String brokerId, int partitionId, int count, long offset, String encodedPartition) { return "MyCustomFileName-"+topic+partitionId; }