@Test public void testGetKeys() throws InvalidProtocolBufferException { DataMessage m1 = prepareSensorisMessage(now, 53.38217, 14.04649); DataMessage m2 = prepareSensorisMessage(now, 53.39363, 14.07643); DataMessages messages = DataMessages.newBuilder().addDataMessage(m1).addDataMessage(m2).build(); Map<MetadataName, String> metadata = new HashMap<>(); metadata.put(MetadataName.INGESTION_TIME, String.valueOf(now)); Map<Map<String, Object>, byte[]> indexKeysAndPayloads = example.getSplittedKeys(metadata, messages.toByteArray()); Assert.assertThat(indexKeysAndPayloads.size(), is(2)); Map<String, Object> nowAndWesternTile = new HashMap<>(); nowAndWesternTile.put(SensorisProtobufSplittedUDFExample.TILE_ID, 92259L); nowAndWesternTile.put(SensorisProtobufSplittedUDFExample.EVENT_TIME, now); Map<String, Object> nowAndEasternTile = new HashMap<>(); nowAndEasternTile.put(SensorisProtobufSplittedUDFExample.TILE_ID, 92262L); nowAndEasternTile.put(SensorisProtobufSplittedUDFExample.EVENT_TIME, now); Assert.assertTrue( indexKeysAndPayloads .keySet() .containsAll(Arrays.asList(nowAndWesternTile, nowAndEasternTile))); Assert.assertEquals( m1, DataMessages.parseFrom(indexKeysAndPayloads.get(nowAndWesternTile)).getDataMessage(0)); Assert.assertEquals( m2, DataMessages.parseFrom(indexKeysAndPayloads.get(nowAndEasternTile)).getDataMessage(0)); }
@Override public Map<Map<String, Object>, byte[]> getSplittedKeys( Map<MetadataName, String> map, byte[] bytes) { try { DataMessages messages = DataMessages.parseFrom(bytes); LOG.debug("Processing {} DataMessage objects", messages.getDataMessageCount()); Map<Map<String, Object>, List<DataMessage>> indexedMessages = messages .getDataMessageList() .stream() .filter(message -> message.getEventGroupCount() > 0) .collect( Collectors.toMap( message -> { Map<String, Object> indexAttributes = new HashMap<>(); VehiclePositionAndOrientation firstPositionReport = message .getEventGroup(0) .getLocalizationCategory() .getVehiclePositionAndOrientation(0); indexAttributes.put( EVENT_TIME, firstPositionReport .getEnvelope() .getTimestamp() .getPosixTime() .getValue()); PositionAndAccuracy.Geographic firstReportedGeolocation = firstPositionReport.getPositionAndAccuracy().getGeographicWgs84(); indexAttributes.put( TILE_ID, HERE_TILE_RESOLVER.fromCoordinate( new GeoCoordinate( firstReportedGeolocation.getLatitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION, firstReportedGeolocation.getLongitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION))); return indexAttributes; }, message -> { List<DataMessage> wrapped = new ArrayList<>(); wrapped.add(message); return wrapped; }, (left, right) -> { left.addAll(right); return left; })); if (LOG.isDebugEnabled()) { indexedMessages.forEach( (key, value) -> LOG.debug( "{} messages keyed to {} from the same DataMessages container", value.size(), key)); } return indexedMessages .entrySet() .stream() .collect( Collectors.toMap( Map.Entry::getKey, entry -> DataMessages.newBuilder() .addAllDataMessage(entry.getValue()) .build() .toByteArray())); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing Sensoris DataMessages error", e); } return null; }
SensorisProtobufSplittedUDFExample implements SplittedUDF { @Override public Map<Map<String, Object>, byte[]> getSplittedKeys( Map<MetadataName, String> map, byte[] bytes) { try { DataMessages messages = DataMessages.parseFrom(bytes); LOG.debug("Processing {} DataMessage objects", messages.getDataMessageCount()); Map<Map<String, Object>, List<DataMessage>> indexedMessages = messages .getDataMessageList() .stream() .filter(message -> message.getEventGroupCount() > 0) .collect( Collectors.toMap( message -> { Map<String, Object> indexAttributes = new HashMap<>(); VehiclePositionAndOrientation firstPositionReport = message .getEventGroup(0) .getLocalizationCategory() .getVehiclePositionAndOrientation(0); indexAttributes.put( EVENT_TIME, firstPositionReport .getEnvelope() .getTimestamp() .getPosixTime() .getValue()); PositionAndAccuracy.Geographic firstReportedGeolocation = firstPositionReport.getPositionAndAccuracy().getGeographicWgs84(); indexAttributes.put( TILE_ID, HERE_TILE_RESOLVER.fromCoordinate( new GeoCoordinate( firstReportedGeolocation.getLatitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION, firstReportedGeolocation.getLongitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION))); return indexAttributes; }, message -> { List<DataMessage> wrapped = new ArrayList<>(); wrapped.add(message); return wrapped; }, (left, right) -> { left.addAll(right); return left; })); if (LOG.isDebugEnabled()) { indexedMessages.forEach( (key, value) -> LOG.debug( "{} messages keyed to {} from the same DataMessages container", value.size(), key)); } return indexedMessages .entrySet() .stream() .collect( Collectors.toMap( Map.Entry::getKey, entry -> DataMessages.newBuilder() .addAllDataMessage(entry.getValue()) .build() .toByteArray())); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing Sensoris DataMessages error", e); } return null; } }
SensorisProtobufSplittedUDFExample implements SplittedUDF { @Override public Map<Map<String, Object>, byte[]> getSplittedKeys( Map<MetadataName, String> map, byte[] bytes) { try { DataMessages messages = DataMessages.parseFrom(bytes); LOG.debug("Processing {} DataMessage objects", messages.getDataMessageCount()); Map<Map<String, Object>, List<DataMessage>> indexedMessages = messages .getDataMessageList() .stream() .filter(message -> message.getEventGroupCount() > 0) .collect( Collectors.toMap( message -> { Map<String, Object> indexAttributes = new HashMap<>(); VehiclePositionAndOrientation firstPositionReport = message .getEventGroup(0) .getLocalizationCategory() .getVehiclePositionAndOrientation(0); indexAttributes.put( EVENT_TIME, firstPositionReport .getEnvelope() .getTimestamp() .getPosixTime() .getValue()); PositionAndAccuracy.Geographic firstReportedGeolocation = firstPositionReport.getPositionAndAccuracy().getGeographicWgs84(); indexAttributes.put( TILE_ID, HERE_TILE_RESOLVER.fromCoordinate( new GeoCoordinate( firstReportedGeolocation.getLatitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION, firstReportedGeolocation.getLongitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION))); return indexAttributes; }, message -> { List<DataMessage> wrapped = new ArrayList<>(); wrapped.add(message); return wrapped; }, (left, right) -> { left.addAll(right); return left; })); if (LOG.isDebugEnabled()) { indexedMessages.forEach( (key, value) -> LOG.debug( "{} messages keyed to {} from the same DataMessages container", value.size(), key)); } return indexedMessages .entrySet() .stream() .collect( Collectors.toMap( Map.Entry::getKey, entry -> DataMessages.newBuilder() .addAllDataMessage(entry.getValue()) .build() .toByteArray())); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing Sensoris DataMessages error", e); } return null; }  }
SensorisProtobufSplittedUDFExample implements SplittedUDF { @Override public Map<Map<String, Object>, byte[]> getSplittedKeys( Map<MetadataName, String> map, byte[] bytes) { try { DataMessages messages = DataMessages.parseFrom(bytes); LOG.debug("Processing {} DataMessage objects", messages.getDataMessageCount()); Map<Map<String, Object>, List<DataMessage>> indexedMessages = messages .getDataMessageList() .stream() .filter(message -> message.getEventGroupCount() > 0) .collect( Collectors.toMap( message -> { Map<String, Object> indexAttributes = new HashMap<>(); VehiclePositionAndOrientation firstPositionReport = message .getEventGroup(0) .getLocalizationCategory() .getVehiclePositionAndOrientation(0); indexAttributes.put( EVENT_TIME, firstPositionReport .getEnvelope() .getTimestamp() .getPosixTime() .getValue()); PositionAndAccuracy.Geographic firstReportedGeolocation = firstPositionReport.getPositionAndAccuracy().getGeographicWgs84(); indexAttributes.put( TILE_ID, HERE_TILE_RESOLVER.fromCoordinate( new GeoCoordinate( firstReportedGeolocation.getLatitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION, firstReportedGeolocation.getLongitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION))); return indexAttributes; }, message -> { List<DataMessage> wrapped = new ArrayList<>(); wrapped.add(message); return wrapped; }, (left, right) -> { left.addAll(right); return left; })); if (LOG.isDebugEnabled()) { indexedMessages.forEach( (key, value) -> LOG.debug( "{} messages keyed to {} from the same DataMessages container", value.size(), key)); } return indexedMessages .entrySet() .stream() .collect( Collectors.toMap( Map.Entry::getKey, entry -> DataMessages.newBuilder() .addAllDataMessage(entry.getValue()) .build() .toByteArray())); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing Sensoris DataMessages error", e); } return null; }  @Override Map<Map<String, Object>, byte[]> getSplittedKeys(
      Map<MetadataName, String> map, byte[] bytes); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files); }
SensorisProtobufSplittedUDFExample implements SplittedUDF { @Override public Map<Map<String, Object>, byte[]> getSplittedKeys( Map<MetadataName, String> map, byte[] bytes) { try { DataMessages messages = DataMessages.parseFrom(bytes); LOG.debug("Processing {} DataMessage objects", messages.getDataMessageCount()); Map<Map<String, Object>, List<DataMessage>> indexedMessages = messages .getDataMessageList() .stream() .filter(message -> message.getEventGroupCount() > 0) .collect( Collectors.toMap( message -> { Map<String, Object> indexAttributes = new HashMap<>(); VehiclePositionAndOrientation firstPositionReport = message .getEventGroup(0) .getLocalizationCategory() .getVehiclePositionAndOrientation(0); indexAttributes.put( EVENT_TIME, firstPositionReport .getEnvelope() .getTimestamp() .getPosixTime() .getValue()); PositionAndAccuracy.Geographic firstReportedGeolocation = firstPositionReport.getPositionAndAccuracy().getGeographicWgs84(); indexAttributes.put( TILE_ID, HERE_TILE_RESOLVER.fromCoordinate( new GeoCoordinate( firstReportedGeolocation.getLatitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION, firstReportedGeolocation.getLongitude().getValue() / SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION))); return indexAttributes; }, message -> { List<DataMessage> wrapped = new ArrayList<>(); wrapped.add(message); return wrapped; }, (left, right) -> { left.addAll(right); return left; })); if (LOG.isDebugEnabled()) { indexedMessages.forEach( (key, value) -> LOG.debug( "{} messages keyed to {} from the same DataMessages container", value.size(), key)); } return indexedMessages .entrySet() .stream() .collect( Collectors.toMap( Map.Entry::getKey, entry -> DataMessages.newBuilder() .addAllDataMessage(entry.getValue()) .build() .toByteArray())); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing Sensoris DataMessages error", e); } return null; }  @Override Map<Map<String, Object>, byte[]> getSplittedKeys(
      Map<MetadataName, String> map, byte[] bytes); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files); static final double SENSORIS_GEOGRAPHIC_SURFACE_RESOLUTION; static final double SENSORIS_GEOGRAPHIC_ALTITUDE_RESOLUTION; }