@Test public void testGetKeys() { long timestamp = ZonedDateTime.now(ZoneOffset.UTC).truncatedTo(ChronoUnit.HOURS).toInstant().toEpochMilli(); double longitude = 10d; double latitude = 10d; Calendar cal = Calendar.getInstance(TimeZone.getTimeZone("UTC")); cal.setTimeInMillis(timestamp); SdiiMessage.Message sdiiMessage = prepareSDIIMessage(timestamp, true, longitude, latitude, "version1"); Map<MetadataName, String> metadata = new HashMap<>(); metadata.put(MetadataName.INGESTION_TIME, String.valueOf(timestamp)); Map<String, Object> keys = example.getKeys(metadata, sdiiMessage.toByteArray()); Assert.assertEquals( HereQuadFactory.INSTANCE .getMapQuadByLocation(latitude, longitude, ParquetSimpleKeyExample.ZOOM_LEVEL) .getLongKey(), keys.get(ParquetSimpleKeyExample.TILE_ID)); Assert.assertEquals(timestamp, keys.get(ParquetSimpleKeyExample.INGESTION_TIME)); Assert.assertEquals( SdiiCommon.SignRecognition.getDescriptor().getName(), keys.get(ParquetSimpleKeyExample.EVENT_TYPE)); }
@Override public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload) { SdiiMessage.Message sdiiMessage; try { sdiiMessage = SdiiMessage.Message.parseFrom(payload); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing SDII Message error", e); return null; } Map<String, Object> result = new HashMap<>(); if (sdiiMessage.getPath() != null && sdiiMessage.getPath().getPositionEstimateList() != null && !sdiiMessage.getPath().getPositionEstimateList().isEmpty()) { TreeSet<SdiiCommon.PositionEstimate> positionEstimates = new TreeSet<>(Comparator.comparingLong(SdiiCommon.PositionEstimate::getTimeStampUTCMs)); positionEstimates.addAll(sdiiMessage.getPath().getPositionEstimateList()); result.put( TILE_ID, HereQuadFactory.INSTANCE .getMapQuadByLocation( positionEstimates.first().getLatitudeDeg(), positionEstimates.first().getLongitudeDeg(), ZOOM_LEVEL) .getLongKey()); } else { result.put(TILE_ID, null); } result.put(INGESTION_TIME, Long.parseLong(metadata.get(MetadataName.INGESTION_TIME))); if (!sdiiMessage.hasPathEvents()) { result.put(EVENT_TYPE, PROBE); } else { String event = "unknown"; List<SdiiCommon.SignRecognition> signRecognitionList = sdiiMessage.getPathEvents().getSignRecognitionList(); if (!signRecognitionList.isEmpty()) { event = SdiiCommon.SignRecognition.getDescriptor().getName(); } result.put(EVENT_TYPE, event); } return result; }
ParquetSimpleKeyExample implements SimpleUDF { @Override public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload) { SdiiMessage.Message sdiiMessage; try { sdiiMessage = SdiiMessage.Message.parseFrom(payload); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing SDII Message error", e); return null; } Map<String, Object> result = new HashMap<>(); if (sdiiMessage.getPath() != null && sdiiMessage.getPath().getPositionEstimateList() != null && !sdiiMessage.getPath().getPositionEstimateList().isEmpty()) { TreeSet<SdiiCommon.PositionEstimate> positionEstimates = new TreeSet<>(Comparator.comparingLong(SdiiCommon.PositionEstimate::getTimeStampUTCMs)); positionEstimates.addAll(sdiiMessage.getPath().getPositionEstimateList()); result.put( TILE_ID, HereQuadFactory.INSTANCE .getMapQuadByLocation( positionEstimates.first().getLatitudeDeg(), positionEstimates.first().getLongitudeDeg(), ZOOM_LEVEL) .getLongKey()); } else { result.put(TILE_ID, null); } result.put(INGESTION_TIME, Long.parseLong(metadata.get(MetadataName.INGESTION_TIME))); if (!sdiiMessage.hasPathEvents()) { result.put(EVENT_TYPE, PROBE); } else { String event = "unknown"; List<SdiiCommon.SignRecognition> signRecognitionList = sdiiMessage.getPathEvents().getSignRecognitionList(); if (!signRecognitionList.isEmpty()) { event = SdiiCommon.SignRecognition.getDescriptor().getName(); } result.put(EVENT_TYPE, event); } return result; } }
ParquetSimpleKeyExample implements SimpleUDF { @Override public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload) { SdiiMessage.Message sdiiMessage; try { sdiiMessage = SdiiMessage.Message.parseFrom(payload); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing SDII Message error", e); return null; } Map<String, Object> result = new HashMap<>(); if (sdiiMessage.getPath() != null && sdiiMessage.getPath().getPositionEstimateList() != null && !sdiiMessage.getPath().getPositionEstimateList().isEmpty()) { TreeSet<SdiiCommon.PositionEstimate> positionEstimates = new TreeSet<>(Comparator.comparingLong(SdiiCommon.PositionEstimate::getTimeStampUTCMs)); positionEstimates.addAll(sdiiMessage.getPath().getPositionEstimateList()); result.put( TILE_ID, HereQuadFactory.INSTANCE .getMapQuadByLocation( positionEstimates.first().getLatitudeDeg(), positionEstimates.first().getLongitudeDeg(), ZOOM_LEVEL) .getLongKey()); } else { result.put(TILE_ID, null); } result.put(INGESTION_TIME, Long.parseLong(metadata.get(MetadataName.INGESTION_TIME))); if (!sdiiMessage.hasPathEvents()) { result.put(EVENT_TYPE, PROBE); } else { String event = "unknown"; List<SdiiCommon.SignRecognition> signRecognitionList = sdiiMessage.getPathEvents().getSignRecognitionList(); if (!signRecognitionList.isEmpty()) { event = SdiiCommon.SignRecognition.getDescriptor().getName(); } result.put(EVENT_TYPE, event); } return result; }  }
ParquetSimpleKeyExample implements SimpleUDF { @Override public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload) { SdiiMessage.Message sdiiMessage; try { sdiiMessage = SdiiMessage.Message.parseFrom(payload); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing SDII Message error", e); return null; } Map<String, Object> result = new HashMap<>(); if (sdiiMessage.getPath() != null && sdiiMessage.getPath().getPositionEstimateList() != null && !sdiiMessage.getPath().getPositionEstimateList().isEmpty()) { TreeSet<SdiiCommon.PositionEstimate> positionEstimates = new TreeSet<>(Comparator.comparingLong(SdiiCommon.PositionEstimate::getTimeStampUTCMs)); positionEstimates.addAll(sdiiMessage.getPath().getPositionEstimateList()); result.put( TILE_ID, HereQuadFactory.INSTANCE .getMapQuadByLocation( positionEstimates.first().getLatitudeDeg(), positionEstimates.first().getLongitudeDeg(), ZOOM_LEVEL) .getLongKey()); } else { result.put(TILE_ID, null); } result.put(INGESTION_TIME, Long.parseLong(metadata.get(MetadataName.INGESTION_TIME))); if (!sdiiMessage.hasPathEvents()) { result.put(EVENT_TYPE, PROBE); } else { String event = "unknown"; List<SdiiCommon.SignRecognition> signRecognitionList = sdiiMessage.getPathEvents().getSignRecognitionList(); if (!signRecognitionList.isEmpty()) { event = SdiiCommon.SignRecognition.getDescriptor().getName(); } result.put(EVENT_TYPE, event); } return result; }  @Override Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files); }
ParquetSimpleKeyExample implements SimpleUDF { @Override public Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload) { SdiiMessage.Message sdiiMessage; try { sdiiMessage = SdiiMessage.Message.parseFrom(payload); } catch (InvalidProtocolBufferException e) { LOG.error("Parsing SDII Message error", e); return null; } Map<String, Object> result = new HashMap<>(); if (sdiiMessage.getPath() != null && sdiiMessage.getPath().getPositionEstimateList() != null && !sdiiMessage.getPath().getPositionEstimateList().isEmpty()) { TreeSet<SdiiCommon.PositionEstimate> positionEstimates = new TreeSet<>(Comparator.comparingLong(SdiiCommon.PositionEstimate::getTimeStampUTCMs)); positionEstimates.addAll(sdiiMessage.getPath().getPositionEstimateList()); result.put( TILE_ID, HereQuadFactory.INSTANCE .getMapQuadByLocation( positionEstimates.first().getLatitudeDeg(), positionEstimates.first().getLongitudeDeg(), ZOOM_LEVEL) .getLongKey()); } else { result.put(TILE_ID, null); } result.put(INGESTION_TIME, Long.parseLong(metadata.get(MetadataName.INGESTION_TIME))); if (!sdiiMessage.hasPathEvents()) { result.put(EVENT_TYPE, PROBE); } else { String event = "unknown"; List<SdiiCommon.SignRecognition> signRecognitionList = sdiiMessage.getPathEvents().getSignRecognitionList(); if (!signRecognitionList.isEmpty()) { event = SdiiCommon.SignRecognition.getDescriptor().getName(); } result.put(EVENT_TYPE, event); } return result; }  @Override Map<String, Object> getKeys(Map<MetadataName, String> metadata, byte[] payload); @Override byte[] aggregate(Map<String, Object> keys, Iterator<byte[]> messages); @Override byte[] merge(Map<String, Object> keys, Iterator<byte[]> files);  }