SAMZA-2041: add hdfs and kinesis descriptor #857
Conversation
|
@xinyuiscool @atoomula can you take a look? |
|
@dxichen Can you take a look as well? |
| public Map<String, String> toConfig() { | ||
| final Map<String, String> config = new HashMap<>(super.toConfig()); | ||
|
|
||
| final String systemName = getSystemName(); |
There was a problem hiding this comment.
nit: The rest of the code-base does not use final for local variables since it's a bit boiler-platey.
There was a problem hiding this comment.
@vjagadish1989 : using final is good habit and should be encouraged in samza code base. I don't think your suggestion makes sense here.
There was a problem hiding this comment.
Style and good practice discussion will never end. We should talk more about these in team meeting and agree upon things together. I will just go for consistency for now
| * @param valueSerde serde the values in the messages in the stream | ||
| * @param systemDescriptor system descriptor this stream descriptor was obtained from | ||
| */ | ||
| KinesisInputDescriptor(String streamId, Serde valueSerde, SystemDescriptor systemDescriptor) { |
There was a problem hiding this comment.
Should you make the serde typed for KinesisInputDescriptor?
| * | ||
| */ | ||
| public class HdfsInputDescriptor | ||
| extends InputDescriptor<Object, HdfsInputDescriptor> { |
There was a problem hiding this comment.
Any chance we can expose a "typed" API instead of InputDescriptor<Object>?
There was a problem hiding this comment.
Unfortunately, HDFS system is not able to leverage the serde concept in samza. Essentially the data type is determined by file type itself (csv, avro, orc, etc.). Users today need to config different reader to read different types of files. We don't just get bytes and deserialize them.
It may be possible to support a "typed" API after more thoughts and refactor. But it would be beyond the scope of this change
| * @return this input descriptor | ||
| */ | ||
| public KinesisInputDescriptor<StreamMessageType> withRegion(String region) { | ||
| this.region = Optional.of(region); |
There was a problem hiding this comment.
Nit: For defensive programming, I would prefer String.stripToNull(region); Please update for similar invocations.
| .withRegion("STREAM-REGION") | ||
| .withAccessKey("YOUR-ACCESS_KEY") | ||
| .withSecretKey("YOUR-SECRET-KEY") | ||
| .withKCLConfig(kclConfig); | ||
| {% endhighlight %} | ||
|
|
||
| As an example, the below configuration is equivalent to invoking `kclClient#WithTableName(myTable)` on the KCL instance. |
There was a problem hiding this comment.
Is there a way that we may truncate these 2 code examples since they only differ by one line? IE addkclConfig.put("TableName", "myTable"); to the first sample.
There was a problem hiding this comment.
It's not completely straightforward to do so. But I have tried to reduce the duplicate codes to make it less verbose.
| private Optional<String> writeCompressionType = Optional.empty(); | ||
| private Optional<String> writerClass = Optional.empty(); | ||
|
|
||
| private Optional<Long> consumerBufferCapacity = Optional.empty(); |
There was a problem hiding this comment.
I would prefer if we add all the consumer/producer configs in the the input and output descriptor respectively instead of the system descriptor.
There was a problem hiding this comment.
This is another unfortunate unresolved issue in HDFS. Where all the configs are per system not per stream. Once we move to per stream config then it definitely makes sense to place them respectively in input and output descriptor.
|
@dxichen @vjagadish1989 addressed or responded to all your feedback. let me know if you have more comments |
Author: Hai Lu <halu@linkedin.com> Reviewers: Xinyu Liu <xiliu@linkedin.com> Closes apache#857 from lhaiesp/master
No description provided.