forked from feast-dev/feast
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use bzip2 compressed feature set json as pipeline option (feast-dev#466)
* Use bzip2 compressed feature set json as pipeline option * Make decompressor and compressor more generic and extensible * Avoid code duplication in test
- Loading branch information
1 parent
85f398e
commit f12f55c
Showing
22 changed files
with
609 additions
and
80 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
47 changes: 47 additions & 0 deletions
47
core/src/main/java/feast/core/job/option/FeatureSetJsonByteConverter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * Copyright 2018-2020 The Feast Authors | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package feast.core.job.option; | ||
|
|
||
| import com.google.protobuf.InvalidProtocolBufferException; | ||
| import com.google.protobuf.util.JsonFormat; | ||
| import feast.core.FeatureSetProto; | ||
| import feast.ingestion.options.OptionByteConverter; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
|
|
||
| public class FeatureSetJsonByteConverter | ||
| implements OptionByteConverter<List<FeatureSetProto.FeatureSet>> { | ||
|
|
||
| /** | ||
| * Convert list of feature sets to json strings joined by new line, represented as byte arrays | ||
| * | ||
| * @param featureSets List of feature set protobufs | ||
| * @return Byte array representation of the json strings | ||
| * @throws InvalidProtocolBufferException | ||
| */ | ||
| @Override | ||
| public byte[] toByte(List<FeatureSetProto.FeatureSet> featureSets) | ||
| throws InvalidProtocolBufferException { | ||
| JsonFormat.Printer printer = | ||
| JsonFormat.printer().omittingInsignificantWhitespace().printingEnumsAsInts(); | ||
| List<String> featureSetsJson = new ArrayList<>(); | ||
| for (FeatureSetProto.FeatureSet featureSet : featureSets) { | ||
| featureSetsJson.add(printer.print(featureSet.getSpec())); | ||
| } | ||
| return String.join("\n", featureSetsJson).getBytes(); | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
83 changes: 83 additions & 0 deletions
83
core/src/test/java/feast/core/job/option/FeatureSetJsonByteConverterTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,83 @@ | ||
| /* | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| * Copyright 2018-2020 The Feast Authors | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package feast.core.job.option; | ||
|
|
||
| import static org.junit.Assert.*; | ||
|
|
||
| import com.google.protobuf.InvalidProtocolBufferException; | ||
| import feast.core.FeatureSetProto; | ||
| import feast.core.SourceProto; | ||
| import feast.types.ValueProto; | ||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
| import java.util.stream.IntStream; | ||
| import org.junit.Test; | ||
|
|
||
| public class FeatureSetJsonByteConverterTest { | ||
|
|
||
| private FeatureSetProto.FeatureSet newFeatureSet(Integer version, Integer numberOfFeatures) { | ||
| List<FeatureSetProto.FeatureSpec> features = | ||
| IntStream.range(1, numberOfFeatures + 1) | ||
| .mapToObj( | ||
| i -> | ||
| FeatureSetProto.FeatureSpec.newBuilder() | ||
| .setValueType(ValueProto.ValueType.Enum.FLOAT) | ||
| .setName("feature".concat(Integer.toString(i))) | ||
| .build()) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| return FeatureSetProto.FeatureSet.newBuilder() | ||
| .setSpec( | ||
| FeatureSetProto.FeatureSetSpec.newBuilder() | ||
| .setSource( | ||
| SourceProto.Source.newBuilder() | ||
| .setType(SourceProto.SourceType.KAFKA) | ||
| .setKafkaSourceConfig( | ||
| SourceProto.KafkaSourceConfig.newBuilder() | ||
| .setBootstrapServers("somebrokers:9092") | ||
| .setTopic("sometopic"))) | ||
| .addAllFeatures(features) | ||
| .setVersion(version) | ||
| .addEntities( | ||
| FeatureSetProto.EntitySpec.newBuilder() | ||
| .setName("entity") | ||
| .setValueType(ValueProto.ValueType.Enum.STRING))) | ||
| .build(); | ||
| } | ||
|
|
||
| @Test | ||
| public void shouldConvertFeatureSetsAsJsonStringBytes() throws InvalidProtocolBufferException { | ||
| int nrOfFeatureSet = 1; | ||
| int nrOfFeatures = 1; | ||
| List<FeatureSetProto.FeatureSet> featureSets = | ||
| IntStream.range(1, nrOfFeatureSet + 1) | ||
| .mapToObj(i -> newFeatureSet(i, nrOfFeatures)) | ||
| .collect(Collectors.toList()); | ||
|
|
||
| String expectedOutputString = | ||
| "{\"version\":1," | ||
| + "\"entities\":[{\"name\":\"entity\",\"valueType\":2}]," | ||
| + "\"features\":[{\"name\":\"feature1\",\"valueType\":6}]," | ||
| + "\"source\":{" | ||
| + "\"type\":1," | ||
| + "\"kafkaSourceConfig\":{" | ||
| + "\"bootstrapServers\":\"somebrokers:9092\"," | ||
| + "\"topic\":\"sometopic\"}}}"; | ||
| FeatureSetJsonByteConverter byteConverter = new FeatureSetJsonByteConverter(); | ||
| assertEquals(expectedOutputString, new String(byteConverter.toByte(featureSets))); | ||
| } | ||
| } |
Oops, something went wrong.