diff --git a/kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/constant/OdpsSqlParser.java b/kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/constant/OdpsSqlParser.java deleted file mode 100644 index 0650ab67..00000000 --- a/kafka-eagle-common/src/main/java/org/smartloli/kafka/eagle/common/constant/OdpsSqlParser.java +++ /dev/null @@ -1,65 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 - * - * http://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 org.smartloli.kafka.eagle.common.constant; - -import java.util.Collections; -import java.util.List; -import java.util.Map; -import java.util.Set; -import java.util.stream.Collectors; - -import org.smartloli.kafka.eagle.common.util.ThrowExceptionUtils; - -import com.alibaba.druid.sql.SQLUtils; -import com.alibaba.druid.sql.ast.SQLStatement; -import com.alibaba.druid.sql.dialect.odps.visitor.OdpsSchemaStatVisitor; -import com.alibaba.druid.stat.TableStat; -import com.alibaba.druid.stat.TableStat.Name; -import com.alibaba.druid.util.JdbcConstants; - -/** - * The client requests the t operation to parse the SQL and obtain the fields - * and conditions. - * - * @author smartloli. - *

- * Created by May 19, 2019 - */ -public class OdpsSqlParser { - - private OdpsSqlParser() { - } - - /** - * Parser sql mapper kafka tree. - */ - public static Set parserTopic(String sql) { - try { - String dbType = JdbcConstants.MYSQL; - List stmtList = SQLUtils.parseStatements(sql, dbType); - SQLStatement stmt = stmtList.get(0); - OdpsSchemaStatVisitor visitor = new OdpsSchemaStatVisitor(); - stmt.accept(visitor); - Map tabmap = visitor.getTables(); - return tabmap.keySet().stream().map(Name::getName).collect(Collectors.toSet()); - } catch (Exception e) { - ThrowExceptionUtils.print(OdpsSqlParser.class).error("Parser kafka sql has error, msg is ", e); - return Collections.emptySet(); - } - } -} diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java index 459470ce..5f3ff844 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/KafkaServiceImpl.java @@ -33,8 +33,6 @@ import java.util.Properties; import java.util.Set; import java.util.concurrent.TimeUnit; -import java.util.regex.Matcher; -import java.util.regex.Pattern; import javax.management.MBeanServerConnection; import javax.management.ObjectName; @@ -67,7 +65,6 @@ import org.slf4j.LoggerFactory; import org.smartloli.kafka.eagle.common.constant.JmxConstants.BrokerServer; import org.smartloli.kafka.eagle.common.constant.JmxConstants.KafkaServer8; -import org.smartloli.kafka.eagle.common.constant.OdpsSqlParser; import org.smartloli.kafka.eagle.common.protocol.BrokersInfo; import org.smartloli.kafka.eagle.common.protocol.DisplayInfo; import org.smartloli.kafka.eagle.common.protocol.HostsInfo; @@ -90,6 +87,7 @@ import com.alibaba.fastjson.JSONObject; import kafka.zk.KafkaZkClient; +import org.smartloli.kafka.eagle.core.sql.tool.OdpsSqlParser; import scala.Option; import scala.Tuple2; import scala.collection.JavaConversions; @@ -339,16 +337,11 @@ public Map> getConsumers(String clusterAlias, DisplayInfo p /** * According to group, topic and partition to get offset from zookeeper. - * - * @param topic - * Filter topic. - * @param group - * Filter group. - * @param partition - * Filter partition. + * + * @param topic Filter topic. + * @param group Filter group. + * @param partition Filter partition. * @return OffsetZkInfo. - * - * @see org.smartloli.kafka.eagle.domain.OffsetZkInfo */ public OffsetZkInfo getOffset(String clusterAlias, String topic, String group, int partition) { KafkaZkClient zkc = kafkaZKPool.getZkClient(clusterAlias); diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/BrokerServiceImpl.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/BrokerServiceImpl.java index 1ac93a8a..ad9d4453 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/BrokerServiceImpl.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/factory/v2/BrokerServiceImpl.java @@ -17,16 +17,8 @@ */ package org.smartloli.kafka.eagle.core.factory.v2; -import java.util.ArrayList; -import java.util.Comparator; -import java.util.HashMap; -import java.util.HashSet; -import java.util.List; -import java.util.Map; +import java.util.*; import java.util.Map.Entry; -import java.util.Properties; -import java.util.Set; -import java.util.TreeSet; import org.apache.kafka.clients.CommonClientConfigs; import org.apache.kafka.clients.admin.AdminClient; @@ -81,10 +73,20 @@ public long topicNumbers(String clusterAlias) { return topicList(clusterAlias).size(); } - /** Exclude kafka topic(__consumer_offsets). */ - private void excludeTopic(List topics) { - if (topics.contains(Kafka.CONSUMER_OFFSET_TOPIC)) { - topics.remove(Kafka.CONSUMER_OFFSET_TOPIC); + /** + * Exclude kafka topic(__consumer_offsets). + */ + private void excludeTopic(List topics, String clusterAlias, KafkaZkClient zkc) { + + topics.remove(Kafka.CONSUMER_OFFSET_TOPIC); + + Iterator iterator = topics.iterator(); + while (iterator.hasNext()) { + String topic = iterator.next(); + if (!zkc.pathExists(BROKER_TOPICS_PATH + "/" + topic + "/partitions")) { + iterator.remove(); + LOG.info("remove topic" + topic); + } } } @@ -346,7 +348,7 @@ public List topicList(String clusterAlias) { if (zkc.pathExists(BROKER_TOPICS_PATH)) { Seq subBrokerTopicsPaths = zkc.getChildren(BROKER_TOPICS_PATH); topics = JavaConversions.seqAsJavaList(subBrokerTopicsPaths); - excludeTopic(topics); + excludeTopic(topics, clusterAlias,zkc); } } catch (Exception e) { LOG.error("Get topic list has error, msg is " + e.getCause().getMessage()); diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/execute/KafkaSqlParser.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/execute/KafkaSqlParser.java index 1f875a08..d9ba0c5e 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/execute/KafkaSqlParser.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/execute/KafkaSqlParser.java @@ -21,18 +21,13 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.concurrent.ConcurrentMap; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.smartloli.kafka.eagle.common.protocol.KafkaSqlInfo; import org.smartloli.kafka.eagle.core.factory.KafkaFactory; import org.smartloli.kafka.eagle.core.factory.KafkaService; -import org.smartloli.kafka.eagle.core.factory.v2.BrokerFactory; -import org.smartloli.kafka.eagle.core.factory.v2.BrokerService; import org.smartloli.kafka.eagle.core.sql.tool.JSqlUtils; - -import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.google.gson.Gson; @@ -47,7 +42,6 @@ public class KafkaSqlParser { private final static Logger LOG = LoggerFactory.getLogger(KafkaSqlParser.class); private static KafkaService kafkaService = new KafkaFactory().create(); - private static BrokerService brokerService = new BrokerFactory().create(); public static String execute(String clusterAlias, String sql) { JSONObject status = new JSONObject(); diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/function/JSONFunction.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/function/JSONFunction.java index 90ed464d..1296d130 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/function/JSONFunction.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/function/JSONFunction.java @@ -20,6 +20,8 @@ import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; +import java.util.Objects; + /** * Parse a JSONObject or a JSONArray in a kafka topic message using a custom * function. @@ -32,19 +34,31 @@ public class JSONFunction { /** Parse a JSONObject. */ public String JSON(String jsonObject, String key) { - JSONObject object = com.alibaba.fastjson.JSON.parseObject(jsonObject); + JSONObject object = null; + try { + Objects.requireNonNull(jsonObject); + object = com.alibaba.fastjson.JSON.parseObject(jsonObject); + } catch (Exception e) { + return null; + } return object.getString(key); } /** Parse a JSONArray. */ public String JSONS(String jsonArray, String key) { - JSONArray object = com.alibaba.fastjson.JSON.parseArray(jsonArray); - JSONArray target = new JSONArray(); - for (Object tmp : object) { - JSONObject result = (JSONObject) tmp; - JSONObject value = new JSONObject(); - value.put(key, result.getString(key)); - target.add(value); + JSONArray target = null; + try { + Objects.requireNonNull(jsonArray); + JSONArray object = com.alibaba.fastjson.JSON.parseArray(jsonArray); + target = new JSONArray(); + for (Object tmp : object) { + JSONObject result = (JSONObject) tmp; + JSONObject value = new JSONObject(); + value.put(key, result.getString(key)); + target.add(value); + } + } catch (Exception e) { + return null; } return target.toJSONString(); } diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/JSqlUtils.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/JSqlUtils.java index 6e556c3b..9df24e2c 100644 --- a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/JSqlUtils.java +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/JSqlUtils.java @@ -17,21 +17,16 @@ */ package org.smartloli.kafka.eagle.core.sql.tool; -import java.io.File; -import java.io.FileWriter; -import java.io.IOException; -import java.sql.Connection; -import java.sql.DriverManager; -import java.sql.ResultSet; -import java.sql.ResultSetMetaData; -import java.sql.Statement; +import java.sql.*; import java.util.ArrayList; import java.util.HashMap; -import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Properties; +import org.apache.commons.lang3.exception.ExceptionUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.smartloli.kafka.eagle.common.constant.JConstants; import org.smartloli.kafka.eagle.core.sql.common.JSqlMapData; @@ -48,37 +43,56 @@ */ public class JSqlUtils { + private final static Logger LOG = LoggerFactory.getLogger(JSqlUtils.class); + /** - * @param tabSchema : Table column,such as {"id":"integer","name":"varchar"} - * @param dataSets : DataSets ,such as - * [{"id":1,"name":"aaa"},{"id":2,"name":"bbb"},{}...] - * @param sql : such as "SELECT * FROM TBL" - * @return String - * @throws Exception : Throws an exception + * @param tabSchema + * @param dataSets + * @param sql + * @return + * @throws Exception */ public static String query(JSONObject tabSchema, Map>> dataSets, String sql) throws Exception { - String model = createTempJson(); + Connection connection = null; + ResultSet result = null; + List> ret = null; + Statement st = null; + try { + String model = createTempJson(); - JSqlMapData.loadSchema(tabSchema, dataSets); + JSqlMapData.loadSchema(tabSchema, dataSets); - Class.forName(JConstants.KAFKA_DRIVER); - Properties info = new Properties(); - info.setProperty("lex", "JAVA"); + Class.forName(JConstants.KAFKA_DRIVER); + Properties info = new Properties(); + info.setProperty("lex", "JAVA"); - Connection connection = DriverManager.getConnection("jdbc:calcite:model=inline:" + model, info); - Statement st = connection.createStatement(); - ResultSet result = st.executeQuery(sql); - ResultSetMetaData rsmd = result.getMetaData(); - List> ret = new ArrayList>(); - while (result.next()) { - Map map = new HashMap(); - for (int i = 1; i <= rsmd.getColumnCount(); i++) { - map.put(rsmd.getColumnName(i), result.getString(rsmd.getColumnName(i))); + connection = DriverManager.getConnection("jdbc:calcite:model=inline:" + model, info); + st = connection.createStatement(); + result = st.executeQuery(sql); + ResultSetMetaData rsmd = result.getMetaData(); + ret = new ArrayList>(); + while (result.next()) { + Map map = new HashMap(); + for (int i = 1; i <= rsmd.getColumnCount(); i++) { + map.put(rsmd.getColumnName(i), result.getString(rsmd.getColumnName(i))); + } + ret.add(map); + } + } catch (ClassNotFoundException e) { + LOG.error(ExceptionUtils.getStackTrace(e)); + } catch (SQLException e) { + LOG.error(ExceptionUtils.getStackTrace(e)); + } finally { + if (st != null) { + st.close(); + } + if (result != null) { + result.close(); + } + if (connection != null) { + connection.close(); } - ret.add(map); } - result.close(); - connection.close(); return new Gson().toJson(ret); } @@ -100,7 +114,7 @@ public static String toJSONObject(List dataSets) { return new Gson().toJson(results); } - private static String createTempJson() throws IOException { + private static String createTempJson() { JSONObject object = new JSONObject(); object.put("version", "1.21.0"); object.put("defaultSchema", "db"); diff --git a/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/OdpsSqlParser.java b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/OdpsSqlParser.java new file mode 100644 index 00000000..32ae90c3 --- /dev/null +++ b/kafka-eagle-core/src/main/java/org/smartloli/kafka/eagle/core/sql/tool/OdpsSqlParser.java @@ -0,0 +1,115 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 + * + * http://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 org.smartloli.kafka.eagle.core.sql.tool; + +import com.google.common.collect.Sets; +import org.apache.calcite.config.Lex; +import org.apache.calcite.sql.*; +import org.apache.calcite.sql.parser.SqlParseException; +import org.apache.calcite.sql.parser.SqlParser; +import org.smartloli.kafka.eagle.common.util.ThrowExceptionUtils; + +import java.util.Collections; +import java.util.Set; + +/** + * The client requests the t operation to parse the SQL and obtain the fields + * and conditions. + * + * @author smartloli. + *

+ * Created by May 19, 2019 + */ +public class OdpsSqlParser { + + private OdpsSqlParser() { + } + + /** + * Parser sql mapper kafka tree. + */ + public static Set parserTopic(String sql) { + try { + Set tableNames = Sets.newHashSet(); + SqlParser.Config config = SqlParser.configBuilder().setLex(Lex.JAVA).build(); + SqlParser sqlParser = SqlParser.create(sql, config); + SqlNode sqlNode = sqlParser.parseStmt(); + parseNode(sqlNode, tableNames); + return tableNames; + } catch (Exception e) { + ThrowExceptionUtils.print(OdpsSqlParser.class).error("Parser kafka sql has error, msg is ", e); + return Collections.emptySet(); + } + } + + private static void parseNode(SqlNode sqlNode, Set tableNames) { + SqlKind sqlKind = sqlNode.getKind(); + switch (sqlKind) { + case SELECT: + SqlNode sqlFrom = ((SqlSelect) sqlNode).getFrom(); + if (sqlFrom.getKind() == SqlKind.IDENTIFIER) { + tableNames.add(sqlFrom.toString()); + } else { + parseNode(sqlFrom, tableNames); + } + break; + case JOIN: + SqlNode leftNode = ((SqlJoin) sqlNode).getLeft(); + SqlNode rightNode = ((SqlJoin) sqlNode).getRight(); + if (leftNode.getKind() == SqlKind.IDENTIFIER) { + tableNames.add(leftNode.toString()); + } else { + parseNode(leftNode, tableNames); + } + if (rightNode.getKind() == SqlKind.IDENTIFIER) { + tableNames.add(rightNode.toString()); + } else { + parseNode(rightNode, tableNames); + } + break; + case AS: + SqlNode identifierNode = ((SqlBasicCall) sqlNode).getOperands()[0]; + if (identifierNode.getKind() != SqlKind.IDENTIFIER) { + parseNode(identifierNode, tableNames); + } else { + tableNames.add(identifierNode.toString()); + } + break; + case UNION: + SqlNode unionLeft = ((SqlBasicCall) sqlNode).getOperands()[0]; + SqlNode unionRight = ((SqlBasicCall) sqlNode).getOperands()[1]; + if (unionLeft.getKind() == SqlKind.IDENTIFIER) { + tableNames.add(unionLeft.toString()); + } else { + parseNode(unionLeft, tableNames); + } + if (unionRight.getKind() == SqlKind.IDENTIFIER) { + tableNames.add(unionRight.toString()); + } else { + parseNode(unionRight, tableNames); + } + break; + case ORDER_BY: + SqlOrderBy sqlOrderBy = (SqlOrderBy) sqlNode; + parseNode(sqlOrderBy.query, tableNames); + default: + break; + + } + } +} diff --git a/kafka-eagle-core/src/test/java/org/smartloli/kafka/eagle/ipc/TestKSql.java b/kafka-eagle-core/src/test/java/org/smartloli/kafka/eagle/ipc/TestKSql.java index abb7f610..90317886 100644 --- a/kafka-eagle-core/src/test/java/org/smartloli/kafka/eagle/ipc/TestKSql.java +++ b/kafka-eagle-core/src/test/java/org/smartloli/kafka/eagle/ipc/TestKSql.java @@ -69,9 +69,9 @@ public static void calcite() throws Exception { List dts = new ArrayList<>(); dts.add(dataSets); long start = System.currentTimeMillis(); - String rs = JSqlUtils.query(tabSchema, tableName, dts, sql); +// String rs = JSqlUtils.query(tabSchema, tableName, dts, sql); System.out.println("[Spent] :: " + (System.currentTimeMillis() - start) + "ms"); - System.out.println(rs); +// System.out.println(rs); } } diff --git a/kafka-eagle-web/src/test/java/org/smartloli/kafka/eagle/sql/TestKafkaParser.java b/kafka-eagle-web/src/test/java/org/smartloli/kafka/eagle/sql/TestKafkaParser.java index f7c8b222..39db951e 100644 --- a/kafka-eagle-web/src/test/java/org/smartloli/kafka/eagle/sql/TestKafkaParser.java +++ b/kafka-eagle-web/src/test/java/org/smartloli/kafka/eagle/sql/TestKafkaParser.java @@ -32,7 +32,12 @@ public static void main(String[] args) { // String sql = "SELECT \"partition\", \"offset\",\"msg\" from // \"kv-test2019\" where \"partition\" in (0) and \"offset\"=37445 group // by \"partition\" limit 10"; - String sql = "select * from \"kv-test2019\" where \"partition\" in (0) limit 10"; + // "select * from job limit 10"; + // "select * from job where `partition` in (0,1) limit 10"; + //"select JSON(msg,'name') as name,JSON(msg,'sex') as sex from person where JSON(msg,'name') like 'zhangsan'"; +// "select msg from test where `partition` in ('0','1')" +// "select JSON(p.msg,'name') as name,JSON(p.msg,'sex') as sex,JSON(j.msg,'age') as age from person p left join job j on JSON(p.msg,'name') = JSON(j.msg,'name')" + String sql = "select * from person union select * from job"; String result = KafkaSqlParser.execute("cluster1", sql); System.out.println("result: " + result); }