uramot score 14
2017/01/15 02:46 投稿
| storm-kafka の MultiScheme の作成方法を教えてください |
| storm-kafka の MultiScheme の作り方がわかりません |
| KafkaからStormにデータを流すものを作りたいのですが、Storm側でKafkaから送られてきたデータをDeserializeする方法がわからなかったため、質問させて頂きました。 |
| [ここ](http://storm.apache.org/releases/1.0.2/storm-kafka.html)を参考に、KafkaSpoutを作成し、kafka-console-producer から送られてきたデータを受取り、StringScheme()を使って KafkaSpout() から emit されているところまでは確認することが出来ました。 |
| しかし、 kafka connect jdbc を使って sqlite からデータを送る際に、送られてきたデータが単純な String でないため、 StringScheme() では正しくDeserializeすることが出来ませんでした。 |
| こういう場合は、 Scheme を自分で作るというように書いてあったのですが、 Scheme の deserialize メソッドで byteBuffer を読み込む方法がわかりませんでした。 |
| java初心者なので、初歩的な質問かもしれませんが、ご回答よろしくお願いします！！！ |
| ```sql |
| create table example( |
| id int primary key autoincrement not null, |
| text varchar |
| ); |
| ``` |
| ```java |
| public static class SpoutScheme implements Scheme { |
| private static final Charset UTF8_CHARSET = StandardCharsets.UTF_8; |
| public List<Object> deserialize(ByteBuffer bytes) { |
| // ここの bytes を OutputFields に合うようにパースしたいです！ |
| return new Values(deserializeString(bytes)); |
| } |
| public static String deserializeString(ByteBuffer string) { |
| if (string.hasArray()) { |
| int base = string.arrayOffset(); |
| return new String(string.array(), base + string.position(), string.remaining()); |
| } else { |
| return new String(Utils.toByteArray(string), UTF8_CHARSET); |
| } |
| } |
| public Fields getOutputFields() { |
| return new Fields("id", "url"); |
| } |
| ``` |
| ### 環境 |
| centos 6 |
| java 1.8 |
| storm 1.0.2 |
| kafka_2.10 |
uramot score 14
2017/01/15 02:45 投稿
| storm-kafka の MultiScheme の作成方法を教えてください |
| KafkaからStormにデータを流すものを作りたいのですが、Storm側でKafkaから送られてきたデータをDeserializeする方法がわからなかったため、質問させて頂きました。 |
| [ここ](http://storm.apache.org/releases/1.0.2/storm-kafka.html)を参考に、KafkaSpoutを作成し、kafka-console-producer から送られてきたデータを受取り、StringScheme()を使って KafkaSpout() から emit されているところまでは確認することが出来ました。 |
| しかし、 kafka connect jdbc を使って sqlite からデータを送る際に、送られてきたデータが単純な String でないため、 StringScheme() では正しくDeserializeすることが出来ませんでした。 |
| こういう場合は、 Scheme を自分で作るというように書いてあったのですが、 Scheme の deserialize メソッドで byteBuffer を読み込む方法がわかりませんでした。 |
| java初心者なので、初歩的な質問かもしれませんが、ご回答よろしくお願いします！！！ |
| ```sql |
| create table example( |
| id int primary key autoincrement not null, |
| text varchar |
| ); |
| ``` |
| ```java |
| public static class SpoutScheme implements Scheme { |
| private static final Charset UTF8_CHARSET = StandardCharsets.UTF_8; |
| public List<Object> deserialize(ByteBuffer bytes) { |
| // ここの bytes を OutputFields に合うようにパースしたいです！ |
| return new Values(deserializeString(bytes)); |
| } |
| public static String deserializeString(ByteBuffer string) { |
| if (string.hasArray()) { |
| int base = string.arrayOffset(); |
| return new String(string.array(), base + string.position(), string.remaining()); |
| } else { |
| return new String(Utils.toByteArray(string), UTF8_CHARSET); |
| } |
| } |
| public Fields getOutputFields() { |
| return new Fields("id", "url"); |
| } |
| ``` |
| ``` |
| ### 環境 |
| centos 6 |
| java 1.8 |
| storm 1.0.2 |
| kafka_2.10 |
