scheme:将kafka传到spout里的数据格式进行转化. record->tuple
mapper:将storm传到kafka的数据格式进行转化.tuple->record
二.为什么要自定义消息格式在很多需求里, 从kafka传递过来的数据并不是单纯的string, 可以是任意对象.当我们需要根据对象的某个属性进行分组时, 默认的new Fields("bytes")就不太适合.但是消息传递的形式还是string.我们可以在传入kafka之前使用fastJson的转化方法将实体对象转化成jsonString.
到了scheme在转换成实体类对象.
三.怎么更改scheme构建kafkaSpout时我们要配置很多参数, 可以看一下kafkaConfig代码.
public final BrokerHosts hosts; //用以获取Kafka broker和partition的信息 public final String topic;//从哪个topic读取消息 public final String clientId; // SimpleConsumer所用的client id public int fetchSizeBytes = 1024 * 1024; //发给Kafka的每个FetchRequest中，用此指定想要的response中总的消息的大小 public int socketTimeoutMs = 10000;//与Kafka broker的连接的socket超时时间 public int fetchMaxWait = 10000; //当服务器没有新消息时，消费者会等待这些时间 public int bufferSizeBytes = 1024 * 1024;//SimpleConsumer所使用的SocketChannel的读缓冲区大小 public MultiScheme scheme = new RawMultiScheme();//从Kafka中取出的byte[]，该如何反序列化 public boolean forceFromStart = false;//是否强制从Kafka中offset最小的开始读起 public long startOffsetTime = kafka.api.OffsetRequest.EarliestTime();//从何时的offset时间开始读，默认为最旧的offset public long maxOffsetBehind = Long.MAX_VALUE;//KafkaSpout读取的进度与目标进度相差多少，相差太多，Spout会丢弃中间的消息 public boolean useStartOffsetTimeIfOffsetOutOfRange = true;//如果所请求的offset对应的消息在Kafka中不存在，是否使startOffsetTime