Skip to content

Commit fae8b65

Browse files
pyalexpyalex
authored
Prevent ingestion job config parser from unwanted fieldMapping transformation (#1261)
Signed-off-by: Oleksii Moskalenko <moskalenko.alexey@gmail.com>
1 parent d8a3795 commit fae8b65

2 files changed

Lines changed: 58 additions & 3 deletions

File tree

‎spark/ingestion/src/main/scala/feast/ingestion/IngestionJob.scala‎

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@
1616
*/
1717
package feast.ingestion
1818

19+
import feast.ingestion.utils.JsonUtils
1920
import org.joda.time.DateTime
2021
import org.json4s._
2122
import org.json4s.jackson.JsonMethods.{parse => parseJSON}
@@ -38,13 +39,20 @@ object IngestionJob {
3839
.text("Mode to operate ingestion job (offline or online)")
3940

4041
opt[String](name = "source")
41-
.action((x, c) =>
42-
parseJSON(x).camelizeKeys.extract[Sources] match {
42+
.action((x, c) => {
43+
val json = parseJSON(x)
44+
JsonUtils
45+
.mapFieldWithParent(json) {
46+
case (parent: String, (key: String, v: JValue)) if !parent.equals("field_mapping") =>
47+
JsonUtils.camelize(key) -> v
48+
case (_, x) => x
49+
}
50+
.extract[Sources] match {
4351
case Sources(file: Some[FileSource], _, _) => c.copy(source = file.get)
4452
case Sources(_, bq: Some[BQSource], _) => c.copy(source = bq.get)
4553
case Sources(_, _, kafka: Some[KafkaSource]) => c.copy(source = kafka.get)
4654
}
47-
)
55+
})
4856
.required()
4957
.text("JSON-encoded source object (e.g. {\"kafka\":{\"bootstrapServers\":...}}")
5058

‎spark/ingestion/src/main/scala/feast/ingestion/utils/JsonUtils.scala‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
/*
2+
* SPDX-License-Identifier: Apache-2.0
3+
* Copyright 2018-2021 The Feast Authors
4+
*
5+
* Licensed under the Apache License, Version 2.0 (the "License");
6+
* you may not use this file except in compliance with the License.
7+
* You may obtain a copy of the License at
8+
*
9+
* https://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
package feast.ingestion.utils
18+
19+
import java.util.Locale.ENGLISH
20+
21+
import org.json4s.{JArray, JField, JObject, JValue}
22+
23+
object JsonUtils {
24+
def mapFieldWithParent(jv: JValue)(f: (String, JField) => JField): JValue = {
25+
def rec(v: JValue, parent: String = ""): JValue = v match {
26+
case JObject(l) => JObject(l.map { case (key, va) => f(parent, key -> rec(va, key)) })
27+
case JArray(l) => JArray(l.map(rec(_, parent)))
28+
case x => x
29+
}
30+
rec(jv)
31+
}
32+
33+
def camelize(word: String): String = {
34+
if (word.nonEmpty) {
35+
val w = pascalize(word)
36+
w.substring(0, 1).toLowerCase(ENGLISH) + w.substring(1)
37+
} else {
38+
word
39+
}
40+
}
41+
42+
def pascalize(word: String): String = {
43+
val lst = word.split("_").toList
44+
(lst.headOption.map(s => s.substring(0, 1).toUpperCase(ENGLISH) + s.substring(1)).get ::
45+
lst.tail.map(s => s.substring(0, 1).toUpperCase + s.substring(1))).mkString("")
46+
}
47+
}

0 commit comments

Comments
 (0)