How should I flatten this json to be able to query?

We use Tranquility and Storm to ingest JSON. However, we have a simple JSON that I’m not sure where to write a flattenSpec in Tranquility

Here’s my sample JSON

“topics”: [{
“name”: “NAME1”,
“score”: 1
}, {
“name”: “TOPIC1”,
“score”: 0.8


I would like to be able to query the count of topic “NAME1”. But i’m not sure where to put the JSON flattenSpec in tranquility

Here’s how I build the Beam

CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeper, new BoundedExponentialBackoffRetry(100, 1000, 5));

DruidSpatialDimension latLng = new DruidSpatialDimension() {
    public SpatialDimensionSchema schema() {
        return new SpatialDimensionSchema("latLng", Arrays.asList("lLa", "lLo"));

DruidSpatialDimension mousePosition = new DruidSpatialDimension() {
    public SpatialDimensionSchema schema() {
        return new SpatialDimensionSchema("mousePosition", Arrays.asList("uMx", "uMy"));

DruidDimensions druidDimensions = DruidDimensions.schemalessWithExclusions(Arrays.asList("uID", "uIP", "pHp", "cTi", "lCi", "xID", "sID", "pID"))
        .withSpatialDimensions(Arrays.asList(latLng, mousePosition));

Beam<Map<String, Object>> beam = DruidBeams.builder(new Timestamper<Map<String, Object>>() {
    public DateTime timestamp(Map<String, Object> theMap) {
        return fmt.parseDateTime(theMap.get(timestampField).toString()).withZone(UTC);

        .location(DruidLocation.create("druid/overlord", dataSource))
        .rollup(DruidRollup.create(druidDimensions, aggregators, NONE))
        .timestampSpec(new TimestampSpec(timestampField, "auto", new DateTime()))
        .tuning(ClusteredBeamTuning.create(HOUR, new Period("PT10M"), new Period("PT10M"), partitionNumber, replicationNumber)).buildBeam();"Finished making Druid Bolt");
return beam;