我从kafka读取了一个json,fieldextractionbolt读取到json将数据提取成元组值并传递给cassandrawriterbolt,然后cassandrawriterbolt在cassandra中写入一条记录,将所有元组值写入单独的列。
Kafka上的json消息-
{"pair":"GBPJPY","bid":134.4563,"ask":134.4354}
现场拔出螺栓-
String message = tuple.getStringByField("message");
Map values = new Gson().fromJson(message, Map.class);
basicOutputCollector.emit(new Values(values.get("pair"), values.get("bid"), values.get("ask")));
Cassandra维特博尔特-
return (CassandraWriterBolt) new CassandraWriterBolt(async(simpleQuery("INSERT INTO currency(pair, ask, bid) VALUES (?, ?, ?);").with(fields("pair", "ask", "bid")))
我试着根据这里给出的答案编写一个测试——如何通过编程插入消息来测试storm拓扑的功能
在我的项目中,我在spring配置中定义了所有螺栓、喷口和流。这使得写/读我的拓扑非常容易。我通过从applicationcontext获取bolt、spouts和streambean来构建拓扑。在我的spring配置中,kafkaspout和cassandrawriterbolt是在prod配置文件下定义的,因此它们只能在prod中使用,而在test配置文件下,我为kafkaspout和cassandrawriterbolt定义存根。对于kafkaspout,我使用fixedtouplespout,对于cassandrawriterbolt,我使用testwordcounter。
这是我的测试
@Test
public void testTopology(){
StormTopology topology = SpringBasedTopologyBuilder.getInstance().buildStormTopologyUsingApplicationContext(applicationContext);
TestJob COMPLETE_TOPOLOGY_TESTJOB = (cluster) -> {
MockedSources mocked = new MockedSources();
mocked.addMockData("kafkaSpout",
new Values("{\"pair\":\"GBPJPY\",\"bid\":134.4563,\"ask\":134.4354}"),
new Values("{\"pair\":\"GBPUSD\",\"bid\":1.4563,\"ask\":1.4354}"));
Config topoConf = new Config();
topoConf.setNumWorkers(2);
CompleteTopologyParam ctp = new CompleteTopologyParam();
ctp.setMockedSources(mocked);
ctp.setStormConf(topoConf);
Map<String, List<FixedTuple>> results = Testing.completeTopology(cluster, topology, ctp);
List<List<Object>> cassandraTuples = Testing.readTuples(results, "cassandraWriterBolt");
List<List<Object>> expectedCassandraTuples = Arrays.asList(Arrays.asList("GBPJPY", 1), Arrays.asList("GBPUSD", 1),
Arrays.asList("134.4563", 1), Arrays.asList("1.4563", 1), Arrays.asList("134.4354", 2));
assertTrue(expectedCassandraTuples + " expected, but found " + cassandraTuples,
Testing.multiseteq(expectedCassandraTuples, cassandraTuples));
MkClusterParam param = new MkClusterParam();
param.setSupervisors(4);
Testing.withSimulatedTimeLocalCluster(param, COMPLETE_TOPOLOGY_TESTJOB);
}
@Configuration
@Import(MainApplication.class)
public static class TestConfig
{
@Bean
public IRichSpout kafkaSpout(){
return new FixedTupleSpout(Arrays.asList(new FixedTuple(Arrays.asList("{\"pair\":\"GBPJPY\",\"bid\":134.4563,\"ask\":134.4354"))), new Fields(new String[]{"message"}));
}
@Bean
public IBasicBolt cassandraWriterBolt(){
return new TestWordCounter();
}
}
我得到的结果不是我所期望的。我有以下错误-
java.lang.AssertionError: [[GBPJPY, 1], [GBPUSD, 1], [134.4563, 1], [1.4563, 1], [134.4354, 2]] expected, but found [[GBPJPY, 1], [GBPUSD, 1]]
看起来,testwordcounter只是将第一个值作为元组读取(仅限货币对,跳过bid和ask)。似乎testwordcounter不是一个正确的选择。什么是正确的存根为cassandrawriterbolt,以便我可以Assert,它将收到两个记录英镑日元和英镑美元的另一个与他们的出价和要价以及?
1条答案
按热度按时间sg24os4d1#
Testing.readTuples(results, "cassandraWriterBolt")
将返回“cassandrawriterbolt”发出的元组。这就是你要测试的吗?我想你是在Assert“cassandrawriterbolt”接受了哪些元组,而不是它发出了什么。你可以在这里做两件事。你可以用
readTuples
从发射到cassandra螺栓的螺栓读取,而不是从cassandra螺栓读取。如果您的拓扑结构很简单(例如,没有太多不同的螺栓写入cassandra螺栓),那么这是一个不错的解决方案。更好的解决方案(imo)是编写一个简单的短螺栓来替换
TestWordCounter
. bolt应该做的唯一一件事是接收输入元组,确认它,并在新元组中发出值。那你可以用
readTuples
读取bolt发出的元组,这些元组将与它接收的值相同。