Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[FSTORE-1023] Mix up of subjectId and schemaId #1122

Open
wants to merge 6 commits into
base: master
Choose a base branch
from
Open

[FSTORE-1023] Mix up of subjectId and schemaId #1122

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
7d26561
init
bubriks Oct 4, 2023
98ecfdd
Merge branch 'master' into FSTORE-1023
bubriks Oct 30, 2023
f7c1842
Merge branch 'master' into FSTORE-1023
bubriks Dec 18, 2023
4558e8c
Merge branch 'master' into FSTORE-1023
bubriks Jan 25, 2024
e627a76
merge fix
bubriks Jan 25, 2024
2a38350
Merge branch 'master' into FSTORE-1023
bubriks Feb 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions java/beam/src/main/java/com/logicalclocks/hsfs/beam/engine/BeamProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,8 +79,8 @@ public BeamProducer(String topic, Map<String, String> properties, Schema schema,
headerMap.put("projectId",
String.valueOf(streamFeatureGroup.getFeatureStore().getProjectId()).getBytes(StandardCharsets.UTF_8));
headerMap.put("featureGroupId", String.valueOf(streamFeatureGroup.getId()).getBytes(StandardCharsets.UTF_8));
headerMap.put("subjectId",
String.valueOf(streamFeatureGroup.getSubject().getId()).getBytes(StandardCharsets.UTF_8));
headerMap.put("schemaId",
String.valueOf(streamFeatureGroup.getSubject().getSchemaId()).getBytes(StandardCharsets.UTF_8));
}

@Override
Expand Down
4 changes: 2 additions & 2 deletions java/flink/src/main/java/com/logicalclocks/hsfs/flink/engine/KafkaRecordSerializer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,8 +49,8 @@ public class KafkaRecordSerializer implements KafkaRecordSerializationSchema<Gen
headerMap.put("projectId",
String.valueOf(streamFeatureGroup.getFeatureStore().getProjectId()).getBytes(StandardCharsets.UTF_8));
headerMap.put("featureGroupId", String.valueOf(streamFeatureGroup.getId()).getBytes(StandardCharsets.UTF_8));
headerMap.put("subjectId",
String.valueOf(streamFeatureGroup.getSubject().getId()).getBytes(StandardCharsets.UTF_8));
headerMap.put("schemaId",
String.valueOf(streamFeatureGroup.getSubject().getSchemaId()).getBytes(StandardCharsets.UTF_8));
}

@Override
Expand Down
10 changes: 7 additions & 3 deletions java/hsfs/src/main/java/com/logicalclocks/hsfs/metadata/Subject.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,13 +34,17 @@ public class Subject {

@Getter
@Setter
private String schema;
private String subject;

@Getter
@Setter
private String subject;
private Integer version;

@Getter
@Setter
private Integer version;
private Integer schemaId;

@Getter
@Setter
private String schema;
}
12 changes: 6 additions & 6 deletions java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/SparkEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -543,7 +543,7 @@ public void writeOnlineDataframe(FeatureGroupBase featureGroupBase, Dataset<Row>
byte[] projectId = String.valueOf(featureGroupBase.getFeatureStore().getProjectId())
.getBytes(StandardCharsets.UTF_8);
byte[] featureGroupId = String.valueOf(featureGroupBase.getId()).getBytes(StandardCharsets.UTF_8);
byte[] subjectId = String.valueOf(featureGroupBase.getSubject().getId()).getBytes(StandardCharsets.UTF_8);
byte[] schemaId = String.valueOf(featureGroupBase.getSubject().getSchemaId()).getBytes(StandardCharsets.UTF_8);

onlineFeatureGroupToAvro(featureGroupBase, encodeComplexFeatures(featureGroupBase, dataset))
.withColumn("headers", array(
Expand All @@ -556,8 +556,8 @@ public void writeOnlineDataframe(FeatureGroupBase featureGroupBase, Dataset<Row>
lit(featureGroupId).as("value")
),
struct(
lit("subjectId").as("key"),
lit(subjectId).as("value")
lit("schemaId").as("key"),
lit(schemaId).as("value")
)
))
.write()
Expand All @@ -575,7 +575,7 @@ public <S> StreamingQuery writeStreamDataframe(FeatureGroupBase featureGroupBase
byte[] projectId = String.valueOf(featureGroupBase.getFeatureStore().getProjectId())
.getBytes(StandardCharsets.UTF_8);
byte[] featureGroupId = String.valueOf(featureGroupBase.getId()).getBytes(StandardCharsets.UTF_8);
byte[] subjectId = String.valueOf(featureGroupBase.getSubject().getId()).getBytes(StandardCharsets.UTF_8);
byte[] schemaId = String.valueOf(featureGroupBase.getSubject().getSchemaId()).getBytes(StandardCharsets.UTF_8);

DataStreamWriter<Row> writer =
onlineFeatureGroupToAvro(featureGroupBase, encodeComplexFeatures(featureGroupBase, dataset))
Expand All @@ -589,8 +589,8 @@ public <S> StreamingQuery writeStreamDataframe(FeatureGroupBase featureGroupBase
lit(featureGroupId).as("value")
),
struct(
lit("subjectId").as("key"),
lit(subjectId).as("value")
lit("schemaId").as("key"),
lit(schemaId).as("value")
)
))
.writeStream()
Expand Down
6 changes: 3 additions & 3 deletions ...park/src/main/java/com/logicalclocks/hsfs/spark/engine/hudi/DeltaStreamerKafkaSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,15 +91,15 @@ protected InputBatch<JavaRDD<GenericRecord>> fetchNewData(Option<String> lastChe
if (totalNewMsgs <= 0L) {
return new InputBatch(Option.empty(), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
} else {
JavaRDD<GenericRecord> newDataRdd = this.toRdd(offsetRanges, props.getString(HudiEngine.SUBJECT_ID));
JavaRDD<GenericRecord> newDataRdd = this.toRdd(offsetRanges, props.getString(HudiEngine.SCHEMA_ID));
return new InputBatch(Option.of(newDataRdd), KafkaOffsetGen.CheckpointUtils.offsetsToStr(offsetRanges));
}
}

private JavaRDD<GenericRecord> toRdd(OffsetRange[] offsetRanges, String subjectId) {
private JavaRDD<GenericRecord> toRdd(OffsetRange[] offsetRanges, String schemaId) {
return KafkaUtils.createRDD(this.sparkContext, this.offsetGen.getKafkaParams(), offsetRanges,
LocationStrategies.PreferConsistent())
.filter(obj -> subjectId.equals(getHeader(obj.headers(), "subjectId")))
.filter(obj -> schemaId.equals(getHeader(obj.headers(), "schemaId")))
.map(obj -> (GenericRecord) obj.value());
}

Expand Down
4 changes: 2 additions & 2 deletions java/spark/src/main/java/com/logicalclocks/hsfs/spark/engine/hudi/HudiEngine.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,7 +126,7 @@ public class HudiEngine {
protected static final String SPARK_MASTER = "yarn";
protected static final String PROJECT_ID = "projectId";
protected static final String FEATURE_STORE_NAME = "featureStoreName";
protected static final String SUBJECT_ID = "subjectId";
protected static final String SCHEMA_ID = "schemaId";
protected static final String FEATURE_GROUP_NAME = "featureGroupName";
protected static final String FEATURE_GROUP_VERSION = "featureGroupVersion";
protected static final String FUNCTION_TYPE = "functionType";
Expand Down Expand Up @@ -367,7 +367,7 @@ public void streamToHoodieTable(SparkSession sparkSession, StreamFeatureGroup st
writeOptions);
hudiWriteOpts.put(PROJECT_ID, String.valueOf(streamFeatureGroup.getFeatureStore().getProjectId()));
hudiWriteOpts.put(FEATURE_STORE_NAME, streamFeatureGroup.getFeatureStore().getName());
hudiWriteOpts.put(SUBJECT_ID, String.valueOf(streamFeatureGroup.getSubject().getId()));
hudiWriteOpts.put(SCHEMA_ID, String.valueOf(streamFeatureGroup.getSubject().getSchemaId()));
hudiWriteOpts.put(FEATURE_GROUP_NAME, streamFeatureGroup.getName());
hudiWriteOpts.put(FEATURE_GROUP_VERSION, String.valueOf(streamFeatureGroup.getVersion()));
hudiWriteOpts.put(HUDI_TABLE_NAME, utils.getFgName(streamFeatureGroup));
Expand Down
4 changes: 3 additions & 1 deletion python/hsfs/engine/python.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1091,7 +1091,9 @@ def _kafka_produce(
"utf8"
),
"featureGroupId": str(feature_group._id).encode("utf8"),
"subjectId": str(feature_group.subject["id"]).encode("utf8"),
"schemaId": str(feature_group.subject["schemaId"]).encode(
"utf8"
),
},
)

Expand Down
10 changes: 4 additions & 6 deletions python/hsfs/engine/spark.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -342,7 +342,7 @@ def save_stream_dataframe(

project_id = str(feature_group.feature_store.project_id).encode("utf8")
feature_group_id = str(feature_group._id).encode("utf8")
subject_id = str(feature_group.subject["id"]).encode("utf8")
schema_id = str(feature_group.subject["schemaId"]).encode("utf8")

query = (
serialized_df.withColumn(
Expand All @@ -355,9 +355,7 @@ def save_stream_dataframe(
lit("featureGroupId").alias("key"),
lit(feature_group_id).alias("value"),
),
struct(
lit("subjectId").alias("key"), lit(subject_id).alias("value")
),
struct(lit("schemaId").alias("key"), lit(schema_id).alias("value")),
),
)
.writeStream.outputMode(output_mode)
Expand Down Expand Up @@ -423,7 +421,7 @@ def _save_online_dataframe(self, feature_group, dataframe, write_options):

project_id = str(feature_group.feature_store.project_id).encode("utf8")
feature_group_id = str(feature_group._id).encode("utf8")
subject_id = str(feature_group.subject["id"]).encode("utf8")
schema_id = str(feature_group.subject["schemaId"]).encode("utf8")

serialized_df.withColumn(
"headers",
Expand All @@ -433,7 +431,7 @@ def _save_online_dataframe(self, feature_group, dataframe, write_options):
lit("featureGroupId").alias("key"),
lit(feature_group_id).alias("value"),
),
struct(lit("subjectId").alias("key"), lit(subject_id).alias("value")),
struct(lit("schemaId").alias("key"), lit(schema_id).alias("value")),
),
).write.format(self.KAFKA_FORMAT).options(**write_options).option(
"topic", feature_group._online_topic_name
Expand Down