Skip to content

Commit 7aad9ba

Browse files
dukewy123坤羽
dukewy123
authored and
坤羽
committed
dsl support json get
1 parent a9a891a commit 7aad9ba

7 files changed

Lines changed: 404 additions & 0 deletions

File tree

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/schema/function/BuildInSqlFunctionTable.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,7 @@
7575
import com.antgroup.geaflow.dsl.udf.table.string.Base64Encode;
7676
import com.antgroup.geaflow.dsl.udf.table.string.Concat;
7777
import com.antgroup.geaflow.dsl.udf.table.string.ConcatWS;
78+
import com.antgroup.geaflow.dsl.udf.table.string.GetJsonObject;
7879
import com.antgroup.geaflow.dsl.udf.table.string.Hash;
7980
import com.antgroup.geaflow.dsl.udf.table.string.IndexOf;
8081
import com.antgroup.geaflow.dsl.udf.table.string.Instr;
@@ -176,6 +177,7 @@ public class BuildInSqlFunctionTable extends ListSqlOperatorTable {
176177
.add(GeaFlowFunction.of(Substr.class))
177178
.add(GeaFlowFunction.of(UrlDecode.class))
178179
.add(GeaFlowFunction.of(UrlEncode.class))
180+
.add(GeaFlowFunction.of(GetJsonObject.class))
179181

180182
// udf.table.other
181183
.add(GeaFlowFunction.of(If.class))

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/main/java/com/antgroup/geaflow/dsl/udf/table/string/GetJsonObject.java‎

Lines changed: 265 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,265 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.dsl.udf.table.string;
21+
22+
import com.antgroup.geaflow.dsl.common.function.Description;
23+
import com.antgroup.geaflow.dsl.common.function.UDF;
24+
import com.google.common.collect.Iterators;
25+
import org.codehaus.jackson.JsonFactory;
26+
import org.codehaus.jackson.JsonParser;
27+
import org.codehaus.jackson.map.ObjectMapper;
28+
import org.codehaus.jackson.map.type.TypeFactory;
29+
import org.codehaus.jackson.type.JavaType;
30+
import java.util.ArrayList;
31+
import java.util.Iterator;
32+
import java.util.LinkedHashMap;
33+
import java.util.List;
34+
import java.util.Map;
35+
import java.util.regex.Matcher;
36+
import java.util.regex.Pattern;
37+
38+
@Description(name = "get_json_object", description = "parse string from json string.")
39+
public class GetJsonObject extends UDF {
40+
41+
private static final JsonFactory JSON_FACTORY = new JsonFactory();
42+
static {
43+
JSON_FACTORY.enable(JsonParser.Feature.ALLOW_UNQUOTED_CONTROL_CHARS);
44+
}
45+
46+
private final Pattern patternKey = Pattern.compile("^([a-zA-Z0-9_\\-\\:\\s]+).*");
47+
private final Pattern patternIndex = Pattern.compile("\\[([0-9]+|\\*)\\]");
48+
private final ObjectMapper MAPPER = new ObjectMapper(JSON_FACTORY);
49+
private final JavaType MAP_TYPE = TypeFactory.fromClass(Map.class);
50+
private final JavaType LIST_TYPE = TypeFactory.fromClass(List.class);
51+
52+
private Map<String, Object> extractObjectCache = new HashCache<String, Object>();
53+
private Map<String, String[]> pathExprCache = new HashCache<String, String[]>();
54+
private Map<String, ArrayList<String>> indexListCache = new HashCache<String, ArrayList<String>>();
55+
private Map<String, String> mKeyGroup1Cache = new HashCache<String, String>();
56+
private Map<String, Boolean> mKeyMatchesCache = new HashCache<String, Boolean>();
57+
58+
private transient AddingList jsonList = new AddingList();
59+
60+
static class HashCache<K, V> extends LinkedHashMap<K, V> {
61+
62+
private static final int CACHE_SIZE = 16;
63+
private static final int INIT_SIZE = 32;
64+
private static final float LOAD_FACTOR = 0.6f;
65+
66+
HashCache() {
67+
super(INIT_SIZE, LOAD_FACTOR);
68+
}
69+
70+
private static final long serialVersionUID = 1;
71+
72+
@Override
73+
protected boolean removeEldestEntry(Map.Entry<K, V> eldest) {
74+
return size() > CACHE_SIZE;
75+
}
76+
}
77+
78+
public String eval(String jsonString, String pathString) {
79+
80+
if (pathString != null && !pathString.startsWith("$.")) {
81+
pathString = "$." + pathString;
82+
}
83+
if (jsonString == null || jsonString.isEmpty() || pathString == null
84+
|| pathString.isEmpty() || pathString.charAt(0) != '$') {
85+
return null;
86+
}
87+
88+
int pathExprStart = 1;
89+
boolean isRootArray = false;
90+
91+
if (pathString.length() > 1) {
92+
if (pathString.charAt(1) == '[') {
93+
pathExprStart = 0;
94+
isRootArray = true;
95+
} else if (pathString.charAt(1) == '.') {
96+
isRootArray = pathString.length() > 2 && pathString.charAt(2) == '[';
97+
} else {
98+
return null;
99+
}
100+
}
101+
102+
String[] pathExpr = pathExprCache.get(pathString);
103+
if (pathExpr == null) {
104+
pathExpr = pathString.split("\\.", -1);
105+
pathExprCache.put(pathString, pathExpr);
106+
}
107+
108+
Object extractObject = extractObjectCache.get(jsonString);
109+
if (extractObject == null) {
110+
JavaType javaType = isRootArray ? LIST_TYPE : MAP_TYPE;
111+
try {
112+
extractObject = MAPPER.readValue(jsonString, javaType);
113+
} catch (Exception e) {
114+
return null;
115+
}
116+
extractObjectCache.put(jsonString, extractObject);
117+
}
118+
for (int i = pathExprStart; i < pathExpr.length; i++) {
119+
if (extractObject == null) {
120+
return null;
121+
}
122+
extractObject = extract(extractObject, pathExpr[i], i == pathExprStart && isRootArray);
123+
}
124+
String result = null;
125+
if (extractObject instanceof Map || extractObject instanceof List) {
126+
try {
127+
result = MAPPER.writeValueAsString(extractObject);
128+
} catch (Exception e) {
129+
return null;
130+
}
131+
} else if (extractObject != null) {
132+
result = extractObject.toString();
133+
} else {
134+
return null;
135+
}
136+
return result;
137+
}
138+
139+
private Object extract(Object json, String path, boolean skipMapProc) {
140+
if (!skipMapProc) {
141+
Matcher mKey = null;
142+
Boolean mKeyMatches = mKeyMatchesCache.get(path);
143+
if (mKeyMatches == null) {
144+
mKey = patternKey.matcher(path);
145+
mKeyMatches = mKey.matches() ? Boolean.TRUE : Boolean.FALSE;
146+
mKeyMatchesCache.put(path, mKeyMatches);
147+
}
148+
if (!mKeyMatches.booleanValue()) {
149+
return null;
150+
}
151+
152+
String mKeyGroup1 = mKeyGroup1Cache.get(path);
153+
if (mKeyGroup1 == null) {
154+
if (mKey == null) {
155+
mKey = patternKey.matcher(path);
156+
mKeyMatches = mKey.matches() ? Boolean.TRUE : Boolean.FALSE;
157+
mKeyMatchesCache.put(path, mKeyMatches);
158+
if (!mKeyMatches.booleanValue()) {
159+
return null;
160+
}
161+
}
162+
mKeyGroup1 = mKey.group(1);
163+
mKeyGroup1Cache.put(path, mKeyGroup1);
164+
}
165+
json = extract_json_withkey(json, mKeyGroup1);
166+
}
167+
// Cache indexList
168+
ArrayList<String> indexList = indexListCache.get(path);
169+
if (indexList == null) {
170+
Matcher mIndex = patternIndex.matcher(path);
171+
indexList = new ArrayList<String>();
172+
while (mIndex.find()) {
173+
indexList.add(mIndex.group(1));
174+
}
175+
indexListCache.put(path, indexList);
176+
}
177+
178+
if (indexList.size() > 0) {
179+
json = extract_json_withindex(json, indexList);
180+
}
181+
182+
return json;
183+
}
184+
185+
private static class AddingList extends ArrayList<Object> {
186+
187+
@Override
188+
public Iterator<Object> iterator() {
189+
return Iterators.forArray(toArray());
190+
}
191+
192+
@Override
193+
public void removeRange(int fromIndex, int toIndex) {
194+
super.removeRange(fromIndex, toIndex);
195+
}
196+
}
197+
198+
@SuppressWarnings("unchecked")
199+
private Object extract_json_withindex(Object json, ArrayList<String> indexList) {
200+
201+
jsonList.clear();
202+
jsonList.add(json);
203+
for (String index : indexList) {
204+
int targets = jsonList.size();
205+
if (index.equalsIgnoreCase("*")) {
206+
for (Object array : jsonList) {
207+
if (array instanceof List) {
208+
for (int j = 0; j < ((List<Object>) array).size(); j++) {
209+
jsonList.add(((List<Object>) array).get(j));
210+
}
211+
}
212+
}
213+
} else {
214+
for (Object array : jsonList) {
215+
int indexValue = Integer.parseInt(index);
216+
if (!(array instanceof List)) {
217+
continue;
218+
}
219+
List<Object> list = (List<Object>) array;
220+
if (indexValue >= list.size()) {
221+
continue;
222+
}
223+
jsonList.add(list.get(indexValue));
224+
}
225+
}
226+
if (jsonList.size() == targets) {
227+
return null;
228+
}
229+
jsonList.removeRange(0, targets);
230+
}
231+
if (jsonList.isEmpty()) {
232+
return null;
233+
}
234+
return (jsonList.size() > 1) ? new ArrayList<Object>(jsonList) : jsonList.get(0);
235+
}
236+
237+
@SuppressWarnings("unchecked")
238+
private Object extract_json_withkey(Object json, String path) {
239+
if (json instanceof List) {
240+
List<Object> jsonList = new ArrayList<Object>();
241+
for (int i = 0; i < ((List<Object>) json).size(); i++) {
242+
Object json_elem = ((List<Object>) json).get(i);
243+
Object json_obj = null;
244+
if (json_elem instanceof Map) {
245+
json_obj = ((Map<String, Object>) json_elem).get(path);
246+
} else {
247+
continue;
248+
}
249+
if (json_obj instanceof List) {
250+
for (int j = 0; j < ((List<Object>) json_obj).size(); j++) {
251+
jsonList.add(((List<Object>) json_obj).get(j));
252+
}
253+
} else if (json_obj != null) {
254+
jsonList.add(json_obj);
255+
}
256+
}
257+
return (jsonList.isEmpty()) ? null : jsonList;
258+
} else if (json instanceof Map) {
259+
return ((Map<String, Object>) json).get(path);
260+
} else {
261+
return null;
262+
}
263+
}
264+
265+
}

‎geaflow/geaflow-dsl/geaflow-dsl-plan/src/test/java/com/antgroup/geaflow/dsl/udf/string/UDFGetJsonObjectTest.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,44 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.dsl.udf.string;
21+
22+
import com.antgroup.geaflow.dsl.udf.table.string.GetJsonObject;
23+
import org.testng.annotations.Test;
24+
25+
import static org.testng.Assert.assertEquals;
26+
27+
public class UDFGetJsonObjectTest {
28+
29+
@Test
30+
public void test() {
31+
32+
GetJsonObject udf = new GetJsonObject();
33+
34+
String jsonStr = "{\"name\": \"Bob\", \"age\": 30, \"address\": " +
35+
"{\"city\": \"Los Angeles\", \"zip\": \"90001\"},\"items\": [\"item1\", \"item2\", \"item3\"]}";
36+
37+
assertEquals("Bob", udf.eval(jsonStr, "name"));
38+
assertEquals("30", udf.eval(jsonStr, "$.age"));
39+
assertEquals("Los Angeles", udf.eval(jsonStr, "$.address.city"));
40+
assertEquals("item3", udf.eval(jsonStr, "$.items[2]"));
41+
assertEquals(null, udf.eval(jsonStr, "gender"));
42+
assertEquals(null, udf.eval(jsonStr, "$.items[3]"));
43+
}
44+
}

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/java/com/antgroup/geaflow/dsl/runtime/query/udf/JsonParserTest.java‎

Lines changed: 36 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,36 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package com.antgroup.geaflow.dsl.runtime.query.udf;
21+
22+
import com.antgroup.geaflow.dsl.runtime.query.QueryTester;
23+
import org.testng.annotations.Test;
24+
25+
public class JsonParserTest {
26+
27+
@Test
28+
public void testJsonPathGet() throws Exception {
29+
QueryTester
30+
.build()
31+
.withQueryPath("/query/json_path_get_001.sql")
32+
.execute()
33+
.checkSinkResult();
34+
}
35+
36+
}

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/data/users_json_object.txt‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
1|jim|15|{"name": "Bob", "age": 30, "address": {"city": "Los Angeles", "zip": "90001"},"items": ["word", "size", "metrics"]}
2+
2|kate|16|{"name": "Alen", "age": 20, "address": {"city": "Zhe Jiang", "zip": "310005"},"items": ["red", "green", "yellow"]}
3+
3|same|20|{"name": "John", "age": 25,"items": ["round", "triangle"]}
4+
4|lucy|21|{"name": "Bush", "age": 11, "address": {"city": "Zhe Jiang", "zip": "310006"}}
5+
5|brown|22|{"name": "Ray"}

‎geaflow/geaflow-dsl/geaflow-dsl-runtime/src/test/resources/expect/json_path_get_001.txt‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
Bob|30|Los Angeles|metrics
2+
Alen|20|Zhe Jiang|yellow
3+
John|25|null|null
4+
Bush|11|Zhe Jiang|null
5+
Ray|null|null|null

0 commit comments

Comments
 (0)