You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@griffin.apache.org by gu...@apache.org on 2017/10/12 04:27:19 UTC
incubator-griffin git commit: Document of griffin dsl and sample
Repository: incubator-griffin
Updated Branches:
refs/heads/master 13107d2a8 -> abda11ef2
Document of griffin dsl and sample
Author: Lionel Liu <bh...@163.com>
Closes #132 from bhlx3lyx7/dsl-modify.
Project: http://git-wip-us.apache.org/repos/asf/incubator-griffin/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-griffin/commit/abda11ef
Tree: http://git-wip-us.apache.org/repos/asf/incubator-griffin/tree/abda11ef
Diff: http://git-wip-us.apache.org/repos/asf/incubator-griffin/diff/abda11ef
Branch: refs/heads/master
Commit: abda11ef2b9f14a30fd393e42a0b44edd8048744
Parents: 13107d2
Author: Lionel Liu <bh...@163.com>
Authored: Thu Oct 12 12:27:12 2017 +0800
Committer: Lionel Liu <bh...@163.com>
Committed: Thu Oct 12 12:27:12 2017 +0800
----------------------------------------------------------------------
.gitignore | 1 +
griffin-doc/dsl-guide.md | 110 ++++++++--
griffin-doc/measure-batch-sample.md | 140 ++++++++++++
griffin-doc/measure-configuration-guide.md | 211 +++++++++++++++++++
griffin-doc/measure-demo-docker.md | 63 ++++++
griffin-doc/measure-streaming-sample-old.md | 204 ++++++++++++++++++
griffin-doc/measure-streaming-sample.md | 204 ------------------
.../batch/HiveBatchDataConnector.scala | 32 ++-
.../batch/TextDirBatchDataConnector.scala | 2 +-
.../measure/rule/dsl/parser/BasicParser.scala | 22 +-
.../config-test-accuracy-streaming-multids.json | 2 -
.../config-test-accuracy-streaming.json | 2 -
.../measure/data/connector/ConnectorTest.scala | 26 +++
.../rule/dsl/parser/BasicParserTest.scala | 7 +-
.../core/measure/repo/DataSourceRepo.java | 26 ---
.../griffin/core/measure/repo/RuleRepo.java | 26 ---
16 files changed, 782 insertions(+), 296 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/.gitignore
----------------------------------------------------------------------
diff --git a/.gitignore b/.gitignore
index c9f7b00..9270ccc 100644
--- a/.gitignore
+++ b/.gitignore
@@ -32,6 +32,7 @@ ui/bower_components/*
ui/node_modules/*
ui/debug.log
ui/package-lock.json
+ui/tmp
derby.log
metastore_db
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/griffin-doc/dsl-guide.md
----------------------------------------------------------------------
diff --git a/griffin-doc/dsl-guide.md b/griffin-doc/dsl-guide.md
index 40a37a4..c668cb9 100644
--- a/griffin-doc/dsl-guide.md
+++ b/griffin-doc/dsl-guide.md
@@ -16,6 +16,7 @@ KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
+
# Apache Griffin DSL Guide
Griffin DSL is designed for DQ measurement, as a SQL-like language, trying to describe the DQ domain request.
@@ -32,7 +33,7 @@ Griffin DSL is SQL-like, case insensitive, and easy to learn.
- `null, nan, true, false`
- `not, and, or`
- `in, between, like, is`
-- `as, where, group, by, having, order, desc, asc, limit`
+- `select, from, as, where, group, by, having, order, desc, asc, limit`
### Operators
- `!, &&, ||, =, !=, <, >, <=, >=, <>`
@@ -42,41 +43,41 @@ Griffin DSL is SQL-like, case insensitive, and easy to learn.
### Literals
- **string**: any string surrounded with a pair of " or ', with escape charactor \ if any request.
- e.g. `"test", 'string 1', "hello \" world \" "`
+ e.g. `"test"`, `'string 1'`, `"hello \" world \" "`
- **number**: double or integer number.
- e.g. `123, 33.5`
+ e.g. `123`, `33.5`
- **time**: a integer with unit in a string, will be translated to a integer number in millisecond.
- e.g. `3d, 5h, 4ms`
+ e.g. `3d`, `5h`, `4ms`
- **boolean**: boolean value directly.
- e.g. `true, false`
+ e.g. `true`, `false`
### Selections
- **selection head**: data source name.
- e.g. `source, target`
+ e.g. `source`, `target`, `` `my table name` ``
- **all field selection**: * or with data source name ahead.
- e.g. `*, source.*, target.*`
+ e.g. `*`, `source.*`, `target.*`
- **field selection**: field name or with data source name ahead.
- e.g. `source.age, target.name, user_id`
+ e.g. `source.age`, `target.name`, `user_id`
- **index selection**: interget between square brackets "[]" with field name ahead.
e.g. `source.attributes[3]`
- **function selection**: function name with brackets "()", with field name ahead or not.
- e.g. `count(*), *.count(), source.user_id.count(), max(source.age)`
+ e.g. `count(*)`, `*.count()`, `source.user_id.count()`, `max(source.age)`
- **alias**: declare an alias after a selection.
- e.g. `source.user_id as id, target.user_name as name`
+ e.g. `source.user_id as id`, `target.user_name as name`
### Math expressions
- **math factor**: literal or function or selection or math exression with brackets.
- e.g. `123, max(1, 2, 3, 4), source.age, (source.age + 13)`
+ e.g. `123`, `max(1, 2, 3, 4)`, `source.age`, `(source.age + 13)`
- **unary math expression**: unary math operator with factor.
e.g. `-(100 - source.score)`
- **binary math expression**: math factors with binary math operators.
- e.g. `source.age + 13, score * 2 + ratio`
+ e.g. `source.age + 13`, `score * 2 + ratio`
### Logical expression
- **in**: in clause like sql.
e.g. `source.country in ("USA", "CHN", "RSA")`
- **between**: between clause like sql.
- e.g. `source.age between 3 and 30, source.age between (3, 30)`
+ e.g. `source.age between 3 and 30`, `source.age between (3, 30)`
- **like**: like clause like sql.
e.g. `source.name like "%abc%"`
- **is null**: is null operator like sql.
@@ -86,7 +87,7 @@ Griffin DSL is SQL-like, case insensitive, and easy to learn.
- **logical factor**: math expression or logical expressions above or other logical expressions with brackets.
e.g. `(source.user_id = target.user_id AND source.age > target.age)`
- **unary logical expression**: unary logical operator with factor.
- e.g. `NOT source.has_data`
+ e.g. `NOT source.has_data`, `!(source.age = target.age)`
- **binary logical expression**: logical factors with binary logical operators, including `and`, `or` and comparison operators.
e.g. `source.age = target.age OR source.ticket = target.tck`
@@ -97,7 +98,84 @@ Griffin DSL is SQL-like, case insensitive, and easy to learn.
### Function
- **argument**: expression.
- **function**: function name with arguments between brackets.
- e.g. `max(source.age, target.age), count(*)`
+ e.g. `max(source.age, target.age)`, `count(*)`
### Clause
-- **select clause**:
\ No newline at end of file
+- **select clause**: the result columns like sql select clause, we can ignore the word "select" in Griffin DSL.
+ e.g. `select user_id.count(), age.max() as max`, `source.user_id.count() as cnt, source.age.min()`
+- **from clause**: the table name like sql from clause, in which the data source name must be one of data source names or the output table name of the former rule steps, we can ignore this clause by configoring the data source name.
+ e.g. `from source`, ``from `target` ``
+- **where clause**: the filter condition like sql where clause, optional.
+ e.g. `where source.id = target.id and source.age = target.age`
+- **group-by clause**: like the group-by clause in sql, optional. Optional having clause could be following.
+ e.g. `group by cntry`, `group by gender having count(*) > 50`
+- **order-by clause**: like the order-by clause, optional.
+ e.g. `order by name`, `order by first_name desc, age asc`
+- **limit clause**: like the limit clause in sql, optional.
+ e.g. `limit 5`
+
+### Accuracy Rule
+Accuracy rule expression in Griffin DSL is a logical expression, telling the mapping relation between data sources.
+ e.g. `source.id = target.id and source.name = target.name and source.age between (target.age, target.age + 5)`
+
+### Profiling Rule
+Profiling rule expression in Griffin DSL is a sql-like expression, with select clause ahead, following optional from clause, where clause, group-by clause, order-by clause, limit clause in order.
+ e.g. `source.gender, source.id.count() where source.age > 20 group by source.gender`, `select country, max(age), min(age), count(*) as cnt from source group by country order by cnt desc limit 5`
+
+## Griffin DSL translation to SQL
+Griffin DSL is defined for DQ measurement, to describe DQ domain problem.
+Actually, in Griffin, we get Griffin DSL rules, translate them into spark-sql rules for calculation in spark-sql engine.
+In DQ domain, there're multiple dimensions, we need to translate them in different ways.
+
+### Accuracy
+For accuracy, we need to get the match count between source and target, the rule describes the mapping relation between data sources. Griffin needs to translate the dsl rule into multiple sql rules.
+For example, the dsl rule is `source.id = target.id and source.name = target.name`, which represents the match condition of accuracy. After the translation, the sql rules are as below:
+- **get miss items from source**: `SELECT source.* FROM source LEFT JOIN target ON coalesce(source.id, '') = coalesce(target.id, '') and coalesce(source.name, '') = coalesce(target.name, '') WHERE (NOT (source.id IS NULL AND source.name IS NULL)) AND (target.id IS NULL AND target.name IS NULL)`, save as table `miss_items`.
+- **get miss count**: `SELECT COUNT(*) AS miss FROM miss_items`, save as table `miss_count`.
+- **get total count from source**: `SELECT COUNT(*) AS total FROM source`, save as table `total_count`.
+- **get accuracy metric**: `SELECT miss_count.miss AS miss, total_count.total AS total, (total_count.total - miss_count.miss) AS matched FROM miss_count FULL JOIN total_count`, save as table `accuracy`.
+
+After the translation, the metrics will be persisted in table `accuracy`.
+
+### Profiling
+For profiling, the request is always the aggregation function of data, the rule is mainly the same as sql, but only supporting `select`, `from`, `where`, `group-by`, `having`, `order-by`, `limit` clauses, which can describe most of the profiling requests. If any complicate request, you can use sql rule directly to describe it.
+For example, the dsl rule is `source.cntry, source.id.count(), source.age.max() group by source.cntry`, which represents the profiling requests. After the translation, the sql rule is as below:
+- **profiling sql rule**: `SELECT source.cntry, count(source.id), max(source.age) FROM source GROUP BY source.cntry`, save as table `profiling`.
+
+After the translation, the metrics will be persisted in table `profiling`.
+
+## ALternative Rules
+You can simply use Griffin DSL rule to describe your problem in DQ domain, for some complicate requirement, you can also use some alternative rules supported by Griffin.
+
+### Spark sql
+Griffin supports spark-sql directly, you can write rule in sql like this:
+```
+{
+ "dsl.type": "spark-sql",
+ "name": "source",
+ "rule": "SELECT count(id) AS cnt, max(timestamp) AS fresh_time FROM source"
+}
+```
+Griffin will calculate it in spark-sql engine directly.
+
+### Data frame operation
+Griffin supports some other operations on data frame in spark, like converting json string data frame into extracted data frame with extracted object schema. For example:
+```
+{
+ "dsl.type": "df-opr",
+ "name": "ext_source",
+ "rule": "from_json",
+ "details": {
+ "df.name": "json_source"
+ }
+}
+```
+Griffin will do the operation to extract json strings.
+Actually, you can also extend the df-opr engine and df-opr adaptor in Griffin to support more types of data frame operations.
+
+## Tips
+Griffin engine runs on spark, it might works in two phases, pre-proc phase and run phase.
+- **Pre-proc phase**: Griffin calculates data source directly, to get appropriate data format, as a preparation for DQ calculation. In this phase, you can use df-opr and spark-sql rules.
+After preparation, to support streaming DQ calculation, a timestamp column will be added in each row of data, so the data frame in run phase contains an extra column named "__tmst".
+- **Run phase**: Griffin calculates with prepared data, to get the DQ metrics. In this phase, you can use griffin-dsl, spark-sql rules, and a part of df-opr rules.
+For griffin-dsl rule, griffin translates it into spark-sql rule with a group-by condition for column "__tmst", it's useful for especially streaming DQ calculation. But for spark-sql rule, griffin use it directly, you need to add the "__tmst" column in your spark-sql rule explicitly, or you can't get correct metrics result after calculation.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/griffin-doc/measure-batch-sample.md
----------------------------------------------------------------------
diff --git a/griffin-doc/measure-batch-sample.md b/griffin-doc/measure-batch-sample.md
new file mode 100644
index 0000000..3783f94
--- /dev/null
+++ b/griffin-doc/measure-batch-sample.md
@@ -0,0 +1,140 @@
+<!--
+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.
+-->
+
+# Measure Batch Sample
+Measures consists of batch measure and streaming measure. This document is for the batch measure sample.
+
+## Batch Accuracy Sample
+```
+{
+ "name": "accu_batch",
+
+ "process.type": "batch",
+
+ "data.sources": [
+ {
+ "name": "src",
+ "connectors": [
+ {
+ "type": "avro",
+ "version": "1.7",
+ "config": {
+ "file.name": "users_info_src.avro"
+ }
+ }
+ ]
+ }, {
+ "name": "tgt",
+ "connectors": [
+ {
+ "type": "avro",
+ "version": "1.7",
+ "config": {
+ "file.name": "users_info_target.avro"
+ }
+ }
+ ]
+ }
+ ],
+
+ "evaluateRule": {
+ "rules": [
+ {
+ "dsl.type": "griffin-dsl",
+ "dq.type": "accuracy",
+ "rule": "src.user_id = tgt.user_id AND upper(src.first_name) = upper(tgt.first_name) AND src.last_name = tgt.last_name",
+ "details": {
+ "source": "src",
+ "target": "tgt",
+ "miss.records": {
+ "name": "miss.records",
+ "persist.type": "record"
+ },
+ "accuracy": {
+ "name": "accu",
+ "persist.type": "metric"
+ },
+ "miss": "miss_count",
+ "total": "total_count",
+ "matched": "matched_count"
+ }
+ }
+ ]
+ }
+}
+```
+Above is the configure file of batch accuracy job.
+
+### Data source
+In this sample, we use avro file as source and target.
+
+### Evaluate rule
+In this accuracy sample, the rule describes the match condition: `src.user_id = tgt.user_id AND upper(src.first_name) = upper(tgt.first_name) AND src.last_name = tgt.last_name`.
+The accuracy metrics will be persisted as metric, with miss column named "miss_count", total column named "total_count", matched column named "matched_count".
+The miss records of source will be persisted as record.
+
+## Batch Profiling Sample
+```
+{
+ "name": "prof_batch_test",
+
+ "process.type": "batch",
+
+ "data.sources": [
+ {
+ "name": "source",
+ "connectors": [
+ {
+ "type": "hive",
+ "version": "1.2",
+ "config": {
+ "database": "griffin",
+ "table.name": "demo_src"
+ }
+ }
+ ]
+ }
+ ],
+
+ "evaluateRule": {
+ "rules": [
+ {
+ "dsl.type": "griffin-dsl",
+ "dq.type": "profiling",
+ "rule": "country, country.count() as cnt group by country order by cnt desc limit 3",
+ "details": {
+ "source": "source",
+ "profiling": {
+ "name": "cntry-group",
+ "persist.type": "metric"
+ }
+ }
+ }
+ ]
+ }
+}
+```
+Above is the configure file of batch profiling job.
+
+### Data source
+In this sample, we use hive table as source.
+
+### Evaluate rule
+In this profiling sample, the rule describes the profiling request: `country, country.count() as cnt group by country order by cnt desc limit 3`.
+The profiling metrics will be persisted as metric, listing the most 3 groups of items in same country.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/griffin-doc/measure-configuration-guide.md
----------------------------------------------------------------------
diff --git a/griffin-doc/measure-configuration-guide.md b/griffin-doc/measure-configuration-guide.md
new file mode 100644
index 0000000..0632927
--- /dev/null
+++ b/griffin-doc/measure-configuration-guide.md
@@ -0,0 +1,211 @@
+<!--
+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.
+-->
+
+# Griffin Measure Configuration Guide
+Griffin measure module needs two configuration files to define the parameters of execution, one is for environment, the other is for dq job.
+
+## Environment Parameters
+```
+{
+ "spark": {
+ "log.level": "WARN",
+ "checkpoint.dir": "hdfs:///griffin/streaming/cp",
+ "batch.interval": "5s",
+ "process.interval": "30s",
+ "config": {
+ "spark.task.maxFailures": 5,
+ "spark.streaming.kafkaMaxRatePerPartition": 1000,
+ "spark.streaming.concurrentJobs": 4
+ }
+ },
+
+ "persist": [
+ {
+ "type": "log",
+ "config": {
+ "max.log.lines": 100
+ }
+ }, {
+ "type": "hdfs",
+ "config": {
+ "path": "hdfs:///griffin/streaming/persist",
+ "max.persist.lines": 10000,
+ "max.lines.per.file": 10000
+ }
+ }
+ ],
+
+ "info.cache": [
+ {
+ "type": "zk",
+ "config": {
+ "hosts": "<zookeeper host ip>:2181",
+ "namespace": "griffin/infocache",
+ "lock.path": "lock",
+ "mode": "persist",
+ "init.clear": true,
+ "close.clear": false
+ }
+ }
+ ]
+}
+```
+Above lists environment parameters.
+
+- **spark**: This field configures spark and spark streaming parameters.
+ + log.level: Level of spark log.
+ + checkpoint.dir: Check point directory of spark streaming, for streaming mode.
+ + batch.interval: Interval of dumping streaming data, for streaming mode.
+ + process.interval: Interval of processing dumped streaming data, for streaming mode.
+ + config: Configuration of spark parameters.
+- **persist**: This field configures list of metrics persist parameters, multiple persist ways are supported. Details of persist configuration [here](#persist).
+- **info.cache**: This field configures list of information cache parameters, multiple cache ways are supported. It is only for streaming dq case. Details of info cache configuration [here](#info-cache).
+
+### <a name="persist"></a>Persist
+- **type**: Metrics persist type, "log", "hdfs" and "http".
+- **config**: Configure parameters of each persist type.
+ + log persist
+ * max.log.lines: the max lines of log.
+ + hdfs persist
+ * path: hdfs path to persist metrics
+ * max.persist.lines: the max lines of total persist data.
+ * max.lines.per.file: the max lines of each persist file.
+ + http persist
+ * api: api to submit persist metrics.
+ * method: http method, "post" default.
+
+### <a name="info-cache"></a>Info Cache
+- **type**: Information cache type, "zk" for zookeeper cache.
+- **config**: Configure parameters of info cache type.
+ + zookeeper cache
+ * hosts: zookeeper hosts list as a string, separated by comma.
+ * namespace: namespace of cache info, "" as default.
+ * lock.path: path of lock info, "lock" as default.
+ * mode: create mode of zookeeper node, "persist" as default.
+ * init.clear: clear cache info when initialize, true default.
+ * close.clear: clear cache info when close connection, false default.
+
+## DQ Job Parameters
+```
+{
+ "name": "accu_batch",
+
+ "process.type": "batch",
+
+ "data.sources": [
+ {
+ "name": "src",
+ "connectors": [
+ {
+ "type": "avro",
+ "version": "1.7",
+ "config": {
+ "file.path": "<path>/<to>",
+ "file.name": "<source-file>.avro"
+ }
+ }
+ ]
+ }, {
+ "name": "tgt",
+ "connectors": [
+ {
+ "type": "avro",
+ "version": "1.7",
+ "config": {
+ "file.path": "<path>/<to>",
+ "file.name": "<target-file>.avro"
+ }
+ }
+ ]
+ }
+ ],
+
+ "evaluateRule": {
+ "rules": [
+ {
+ "dsl.type": "griffin-dsl",
+ "dq.type": "accuracy",
+ "rule": "src.user_id = tgt.user_id AND upper(src.first_name) = upper(tgt.first_name) AND src.last_name = tgt.last_name",
+ "details": {
+ "source": "src",
+ "target": "tgt",
+ "miss.records": {
+ "name": "miss.records",
+ "persist.type": "record"
+ },
+ "accuracy": {
+ "name": "accu",
+ "persist.type": "metric"
+ },
+ "miss": "miss_count",
+ "total": "total_count",
+ "matched": "matched_count"
+ }
+ }
+ ]
+ }
+}
+```
+Above lists DQ job configure parameters.
+
+- **name**: Name of DQ job.
+- **process.type**: Process type of DQ job, "batch" or "streaming".
+- **data.sources**: List of data sources in this DQ job.
+ + name: Name of this data source, it should be different from other data sources.
+ + connectors: List of data connectors combined as the same data source. Details of data connector configuration [here](#data-connector).
+- **evaluateRule**: Evaluate rule parameters of this DQ job.
+ + dsl.type: Default dsl type of all the rules.
+ + rules: List of rules, to define every rule step. Details of rule configuration [here](#rule).
+
+### <a name="data-connector"></a>Data Connector
+- **type**: Data connector type, "avro", "hive", "text-dir" for batch mode, "kafka" for streaming mode.
+- **version**: Version string of data connector type.
+- **config**: Configure parameters of each data connector type.
+ + avro data connector
+ * file.path: avro file path, optional, "" as default.
+ * file.name: avro file name.
+ + hive data connector
+ * database: data base name, optional, "default" as default.
+ * table.name: table name.
+ * partitions: partition conditions string, split by ";" and ",", optional.
+ e.g. `dt=20170410, hour=15; dt=20170411, hour=15; dt=20170412, hour=15`
+ + text dir data connector
+ * dir.path: parent directory path.
+ * data.dir.depth: integer, depth of data directories, 0 as default.
+ * success.file: success file name,
+ * done.file:
+
+### <a name="rule"></a>Rule
+- **dsl.type**: Rule dsl type, "spark-sql", "df-opr" and "griffin-dsl".
+- **name** (step information): Result table name of this rule, optional for "griffin-dsl" type.
+- **persist.type** (step information): Persist type of result table, optional for "griffin-dsl" type. Supporting "metric", "record" and "none" type, "metric" type indicates the result will be persisted as metrics, "record" type indicates the result will be persisted as record only, "none" type indicates the result will not be persisted. Default is "none" type.
+- **update.data.source** (step information): If the result table needs to update the data source, this parameter is the data source name, for streaming accuracy case, optional.
+- **dq.type**: DQ type of this rule, only for "griffin-dsl" type, supporting "accuracy" and "profiling".
+- **details**: Details of this rule, optional.
+ + accuracy dq type detail configuration
+ * source: the data source name which as source in accuracy, default is the name of first data source in "data.sources" if not configured.
+ * target: the data source name which as target in accuracy, default is the name of second data source in "data.sources" if not configured.
+ * miss.records: step information of miss records result table step in accuracy.
+ * accuracy: step information of accuracy result table step in accuracy.
+ * miss: alias of miss column in result table.
+ * total: alias of total column in result table.
+ * matched: alias of matched column in result table.
+ + profiling dq type detail configuration
+ * source: the data source name which as source in profiling, default is the name of first data source in "data.sources" if not configured. If the griffin-dsl rule contains from clause, this parameter is ignored.
+ * profiling: step information of profiling result table step in profiling.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/griffin-doc/measure-demo-docker.md
----------------------------------------------------------------------
diff --git a/griffin-doc/measure-demo-docker.md b/griffin-doc/measure-demo-docker.md
new file mode 100644
index 0000000..49040da
--- /dev/null
+++ b/griffin-doc/measure-demo-docker.md
@@ -0,0 +1,63 @@
+<!--
+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.
+-->
+
+# Griffin Measure Demo Docker
+We've prepared a docker for griffin measure demo.
+
+## Preparation
+1. Install docker.
+2. Download docker image. In this image, the environment for measure module has been prepared, including: hadoop, hive, spark, mysql.
+```
+docker pull bhlx3lyx7/griffin_measure_demo:0.0.1
+```
+3. Run docker image.
+```
+docker run -it -h griffin --name griffin_measure_demo -m 8G --memory-swap -1 \
+-p 42122:2122 -p 47077:7077 -p 46066:6066 -p 48088:8088 -p 48040:8040 \
+-p 43306:3306 -p 49000:9000 -p 48042:8042 -p 48080:8080 -p 47017:27017 \
+-p 49083:9083 -p 48998:8998 -p 49200:9200 bhlx3lyx7/griffin_measure_demo:0.0.1
+```
+4. In this docker container, run the prepared demo.
+- **accuracy demo**: This demo is batch accuracy, source data is Hive table "demo_src", target data is Hive table "demo_tgt", metrics will be persisted in `hdfs:///griffin/persist/accu` after calculation.
+ + switch into `job/accu`.
+ ```
+ cd job/accu
+ ```
+ + run the prepared script.
+ ```
+ ./bgwork.sh
+ ```
+ + check job log.
+ ```
+ tail -f accu.log
+ ```
+- **profiling demo**: This demo is batch profiling, source data is Hive table "demo_src", metrics will be persisted in `hdfs:///griffin/persist/prof` after calculation.
+ + switch into `job/prof`.
+ ```
+ cd job/prof
+ ```
+ + run the prepared script.
+ ```
+ ./bgwork.sh
+ ```
+ + check job log.
+ ```
+ tail -f prof.log
+ ```
+5. You can modify the job configuration file `config.json` of the above demos, or create your own data sources, to get more metrics of data.
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/griffin-doc/measure-streaming-sample-old.md
----------------------------------------------------------------------
diff --git a/griffin-doc/measure-streaming-sample-old.md b/griffin-doc/measure-streaming-sample-old.md
new file mode 100644
index 0000000..004ed3b
--- /dev/null
+++ b/griffin-doc/measure-streaming-sample-old.md
@@ -0,0 +1,204 @@
+<!--
+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.
+-->
+# Measure streaming sample
+Measures consists of batch measure and streaming measure. This document is for the streaming measure sample.
+
+### Data source
+At current, we support kafka as streaming data source.
+In this sample, we also need a kafka as data source.
+
+### Measure type
+At current, we support accuracy measure in streaming mode.
+
+### Kafka decoder
+In kafka, data always needs encode and decode, we support String type kafka data currently, you can also implement and use your decoder for kafka case.
+
+### Environment
+For current griffin streaming case, we need some necessary environment dependencies, zookeeper and hdfs.
+We use zookeeper to cache some checkpoint information, it's optional, but we recommend it.
+We use hdfs to save the temporary data, it's also a recommend selection.
+
+### Streaming accuracy result
+The streaming data will be separated into mini-batches of data, for each mini-batch data, there should be an accuracy result. Therefore, the streaming accuracy result should be a bunch of batch accuracy results with timestamp.
+Considering the latency of streaming data, which means the source data and the matching target data will not exactly reach exactly at the same time, we have to accept some delay of data in streaming mode, by holding unmatched data in memory or disk, and try to match them later until the data is out-time.
+
+## How to run streaming sample
+### Environment Preparation
+At first, we need some environment preparation.
+- Zookeeper: Zookeeper 3.4.10
+- Hadoop: Hadoop 2.6
+- Spark: Spark 1.6
+- Kafka: Kafka 0.8
+
+### Data Preparation
+Create two topics in kafka, for source and target data. For example, topic "source" for source data, and topic "target" for target data.
+Streaming data should also be prepared, the format could be json string, for example:
+Source data could be:
+```
+{"name": "kevin", "age": 24}
+{"name": "jason", "age": 25}
+{"name": "jhon", "age": 28}
+{"name": "steve", "age": 31}
+```
+Target data could be:
+```
+{"name": "kevin", "age": 24}
+{"name": "jason", "age": 25}
+{"name": "steve", "age": 20}
+```
+You need to input the source data and target data into these two topics, through console producer might be a good choice for experimental purpose.
+
+### Configuration Preparation
+Two configuration files are required.
+Environment configuration file: env.json
+```
+{
+ "spark": {
+ "log.level": "WARN",
+ "checkpoint.dir": "hdfs:///griffin/streaming/cp",
+ "batch.interval": "5s",
+ "process.interval": "30s",
+ "config": {
+ "spark.task.maxFailures": 5,
+ "spark.streaming.kafkaMaxRatePerPartition": 1000,
+ "spark.streaming.concurrentJobs": 4
+ }
+ },
+
+ "persist": [
+ {
+ "type": "log",
+ "config": {
+ "max.log.lines": 100
+ }
+ }, {
+ "type": "hdfs",
+ "config": {
+ "path": "hdfs:///griffin/streaming/persist",
+ "max.persist.lines": 10000,
+ "max.lines.per.file": 10000
+ }
+ }
+ ],
+
+ "info.cache": [
+ {
+ "type": "zk",
+ "config": {
+ "hosts": "<zookeeper host ip>:2181",
+ "namespace": "griffin/infocache",
+ "lock.path": "lock",
+ "mode": "persist",
+ "init.clear": true,
+ "close.clear": false
+ }
+ }
+ ]
+}
+```
+In env.json, "spark" field configures the spark and spark streaming parameters, "persist" field configures the persist ways, we support "log", "hdfs" and "http" ways at current, "info.cache" field configures the information cache parameters, we support zookeeper only at current.
+
+Process configuration file: config.json
+```
+{
+ "name": "streaming-accu-sample",
+ "type": "accuracy",
+ "process.type": "streaming",
+
+ "source": {
+ "type": "kafka",
+ "version": "0.8",
+ "config": {
+ "kafka.config": {
+ "bootstrap.servers": "<kafka host ip>:9092",
+ "group.id": "group1",
+ "auto.offset.reset": "smallest",
+ "auto.commit.enable": "false"
+ },
+ "topics": "source",
+ "key.type": "java.lang.String",
+ "value.type": "java.lang.String"
+ },
+ "cache": {
+ "type": "text",
+ "config": {
+ "file.path": "hdfs:///griffin/streaming/dump/source",
+ "info.path": "source",
+ "ready.time.interval": "10s",
+ "ready.time.delay": "0"
+ },
+ "time.range": ["-5m", "0"]
+ },
+ "match.once": true
+ },
+
+ "target": {
+ "type": "kafka",
+ "version": "0.8",
+ "config": {
+ "kafka.config": {
+ "bootstrap.servers": "<kafka host ip>:9092",
+ "group.id": "group1",
+ "auto.offset.reset": "smallest",
+ "auto.commit.enable": "false"
+ },
+ "topics": "target",
+ "key.type": "java.lang.String",
+ "value.type": "java.lang.String"
+ },
+ "cache": {
+ "type": "text",
+ "config": {
+ "file.path": "hdfs:///griffin/streaming/dump/target",
+ "info.path": "target",
+ "ready.time.interval": "10s",
+ "ready.time.delay": "0"
+ },
+ "time.range": ["-5m", "0"]
+ },
+ "match.once": false
+ },
+
+ "evaluateRule": {
+ "rules": "$source.json().name = $target.json().name AND $source.json().age = $target.json().age"
+ }
+}
+```
+In config.json, "source" and "target" fields configure the data source parameters.
+The "cache" field in data source configuration represents the temporary data cache way, at current we support "text" and "hive" ways. We recommend "text" way, it only depends on hdfs. "time.range" means that the data older than the lower bound should be considered as out-time, and the out-time data will not be calculated any more.
+"match.once" represents the data from this data source could be matched only once or more times.
+"evaluateRule.rule" configures the match rule between each source and target data.
+
+### Run
+Build the measure package.
+```
+mvn clean install
+```
+Get the measure package ```measure-<version>-incubating-SNAPSHOT.jar```, rename it to ```griffin-measure.jar```.
+Put measure package together with env.json and config.json.
+Run the following command:
+```
+spark-submit --class org.apache.griffin.measure.Application \
+--master yarn-client --queue default \
+griffin-measure.jar \
+env.json config.json local,local
+```
+The first two parameters are the paths of env.json and config.json, the third parameter represents the file system type of the two configuration files, "local" or "hdfs" are both supported.
+
+The spark streaming application will be long-time running, you can get the results of each mini-batch of data, during the run-time, you can also input more data into source and target topics, to check the results of the later mini-batches.
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/griffin-doc/measure-streaming-sample.md
----------------------------------------------------------------------
diff --git a/griffin-doc/measure-streaming-sample.md b/griffin-doc/measure-streaming-sample.md
deleted file mode 100644
index 004ed3b..0000000
--- a/griffin-doc/measure-streaming-sample.md
+++ /dev/null
@@ -1,204 +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.
--->
-# Measure streaming sample
-Measures consists of batch measure and streaming measure. This document is for the streaming measure sample.
-
-### Data source
-At current, we support kafka as streaming data source.
-In this sample, we also need a kafka as data source.
-
-### Measure type
-At current, we support accuracy measure in streaming mode.
-
-### Kafka decoder
-In kafka, data always needs encode and decode, we support String type kafka data currently, you can also implement and use your decoder for kafka case.
-
-### Environment
-For current griffin streaming case, we need some necessary environment dependencies, zookeeper and hdfs.
-We use zookeeper to cache some checkpoint information, it's optional, but we recommend it.
-We use hdfs to save the temporary data, it's also a recommend selection.
-
-### Streaming accuracy result
-The streaming data will be separated into mini-batches of data, for each mini-batch data, there should be an accuracy result. Therefore, the streaming accuracy result should be a bunch of batch accuracy results with timestamp.
-Considering the latency of streaming data, which means the source data and the matching target data will not exactly reach exactly at the same time, we have to accept some delay of data in streaming mode, by holding unmatched data in memory or disk, and try to match them later until the data is out-time.
-
-## How to run streaming sample
-### Environment Preparation
-At first, we need some environment preparation.
-- Zookeeper: Zookeeper 3.4.10
-- Hadoop: Hadoop 2.6
-- Spark: Spark 1.6
-- Kafka: Kafka 0.8
-
-### Data Preparation
-Create two topics in kafka, for source and target data. For example, topic "source" for source data, and topic "target" for target data.
-Streaming data should also be prepared, the format could be json string, for example:
-Source data could be:
-```
-{"name": "kevin", "age": 24}
-{"name": "jason", "age": 25}
-{"name": "jhon", "age": 28}
-{"name": "steve", "age": 31}
-```
-Target data could be:
-```
-{"name": "kevin", "age": 24}
-{"name": "jason", "age": 25}
-{"name": "steve", "age": 20}
-```
-You need to input the source data and target data into these two topics, through console producer might be a good choice for experimental purpose.
-
-### Configuration Preparation
-Two configuration files are required.
-Environment configuration file: env.json
-```
-{
- "spark": {
- "log.level": "WARN",
- "checkpoint.dir": "hdfs:///griffin/streaming/cp",
- "batch.interval": "5s",
- "process.interval": "30s",
- "config": {
- "spark.task.maxFailures": 5,
- "spark.streaming.kafkaMaxRatePerPartition": 1000,
- "spark.streaming.concurrentJobs": 4
- }
- },
-
- "persist": [
- {
- "type": "log",
- "config": {
- "max.log.lines": 100
- }
- }, {
- "type": "hdfs",
- "config": {
- "path": "hdfs:///griffin/streaming/persist",
- "max.persist.lines": 10000,
- "max.lines.per.file": 10000
- }
- }
- ],
-
- "info.cache": [
- {
- "type": "zk",
- "config": {
- "hosts": "<zookeeper host ip>:2181",
- "namespace": "griffin/infocache",
- "lock.path": "lock",
- "mode": "persist",
- "init.clear": true,
- "close.clear": false
- }
- }
- ]
-}
-```
-In env.json, "spark" field configures the spark and spark streaming parameters, "persist" field configures the persist ways, we support "log", "hdfs" and "http" ways at current, "info.cache" field configures the information cache parameters, we support zookeeper only at current.
-
-Process configuration file: config.json
-```
-{
- "name": "streaming-accu-sample",
- "type": "accuracy",
- "process.type": "streaming",
-
- "source": {
- "type": "kafka",
- "version": "0.8",
- "config": {
- "kafka.config": {
- "bootstrap.servers": "<kafka host ip>:9092",
- "group.id": "group1",
- "auto.offset.reset": "smallest",
- "auto.commit.enable": "false"
- },
- "topics": "source",
- "key.type": "java.lang.String",
- "value.type": "java.lang.String"
- },
- "cache": {
- "type": "text",
- "config": {
- "file.path": "hdfs:///griffin/streaming/dump/source",
- "info.path": "source",
- "ready.time.interval": "10s",
- "ready.time.delay": "0"
- },
- "time.range": ["-5m", "0"]
- },
- "match.once": true
- },
-
- "target": {
- "type": "kafka",
- "version": "0.8",
- "config": {
- "kafka.config": {
- "bootstrap.servers": "<kafka host ip>:9092",
- "group.id": "group1",
- "auto.offset.reset": "smallest",
- "auto.commit.enable": "false"
- },
- "topics": "target",
- "key.type": "java.lang.String",
- "value.type": "java.lang.String"
- },
- "cache": {
- "type": "text",
- "config": {
- "file.path": "hdfs:///griffin/streaming/dump/target",
- "info.path": "target",
- "ready.time.interval": "10s",
- "ready.time.delay": "0"
- },
- "time.range": ["-5m", "0"]
- },
- "match.once": false
- },
-
- "evaluateRule": {
- "rules": "$source.json().name = $target.json().name AND $source.json().age = $target.json().age"
- }
-}
-```
-In config.json, "source" and "target" fields configure the data source parameters.
-The "cache" field in data source configuration represents the temporary data cache way, at current we support "text" and "hive" ways. We recommend "text" way, it only depends on hdfs. "time.range" means that the data older than the lower bound should be considered as out-time, and the out-time data will not be calculated any more.
-"match.once" represents the data from this data source could be matched only once or more times.
-"evaluateRule.rule" configures the match rule between each source and target data.
-
-### Run
-Build the measure package.
-```
-mvn clean install
-```
-Get the measure package ```measure-<version>-incubating-SNAPSHOT.jar```, rename it to ```griffin-measure.jar```.
-Put measure package together with env.json and config.json.
-Run the following command:
-```
-spark-submit --class org.apache.griffin.measure.Application \
---master yarn-client --queue default \
-griffin-measure.jar \
-env.json config.json local,local
-```
-The first two parameters are the paths of env.json and config.json, the third parameter represents the file system type of the two configuration files, "local" or "hdfs" are both supported.
-
-The spark streaming application will be long-time running, you can get the results of each mini-batch of data, during the run-time, you can also input more data into source and target topics, to check the results of the later mini-batches.
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/HiveBatchDataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/HiveBatchDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/HiveBatchDataConnector.scala
index 5d80d0e..20c9e24 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/HiveBatchDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/HiveBatchDataConnector.scala
@@ -48,17 +48,28 @@ case class HiveBatchDataConnector(sqlContext: SQLContext, dqEngines: DqEngines,
val partitionsString = config.getString(Partitions, "")
val concreteTableName = s"${database}.${tableName}"
- val partitions = partitionsString.split(";").map(s => s.split(",").map(_.trim))
+// val partitions = partitionsString.split(";").map(s => s.split(",").map(_.trim))
+ val partitions: Array[Array[String]] = partitionsString.split(";").flatMap { s =>
+ val arr = s.trim.split(",").flatMap { t =>
+ t.trim match {
+ case p if (p.nonEmpty) => Some(p)
+ case _ => None
+ }
+ }
+ if (arr.size > 0) Some(arr) else None
+ }
def data(ms: Long): Option[DataFrame] = {
try {
- val df = sqlContext.sql(dataSql)
+ val dtSql = dataSql
+ info(dtSql)
+ val df = sqlContext.sql(dtSql)
val dfOpt = Some(df)
val preDfOpt = preProcess(dfOpt, ms)
preDfOpt
} catch {
case e: Throwable => {
- error(s"load hive table ${concreteTableName} fails")
+ error(s"load hive table ${concreteTableName} fails: ${e.getMessage}")
None
}
}
@@ -131,12 +142,15 @@ case class HiveBatchDataConnector(sqlContext: SQLContext, dqEngines: DqEngines,
}
private def dataSql(): String = {
- val clauses = partitions.map { prtn =>
- val cls = prtn.mkString(" AND ")
- if (cls.isEmpty) s"SELECT * FROM ${concreteTableName}"
- else s"SELECT * FROM ${concreteTableName} WHERE ${cls}"
- }
- clauses.mkString(" UNION ALL ")
+ val tableClause = s"SELECT * FROM ${concreteTableName}"
+ val validPartitions = partitions.filter(_.size > 0)
+ if (validPartitions.size > 0) {
+ val clauses = validPartitions.map { prtn =>
+ val cls = prtn.mkString(" AND ")
+ s"${tableClause} WHERE ${cls}"
+ }
+ clauses.mkString(" UNION ALL ")
+ } else tableClause
}
// private def toTuple[A <: AnyRef](as: Seq[A]): Product = {
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/TextDirBatchDataConnector.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/TextDirBatchDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/TextDirBatchDataConnector.scala
index abc547b..13ffe89 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/TextDirBatchDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/data/connector/batch/TextDirBatchDataConnector.scala
@@ -75,7 +75,7 @@ case class TextDirBatchDataConnector(sqlContext: SQLContext, dqEngines: DqEngine
if (depth <= 0) {
subDirs.filter(filteFunc)
} else {
- listSubDirs(subDirs, depth, filteFunc)
+ listSubDirs(subDirs, depth - 1, filteFunc)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/main/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParser.scala
----------------------------------------------------------------------
diff --git a/measure/src/main/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParser.scala b/measure/src/main/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParser.scala
index f55b1f8..7d9646d 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParser.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParser.scala
@@ -86,7 +86,12 @@ trait BasicParser extends JavaTokenParsers with Serializable {
* <alias-expr> = <expr> <as> <name>
*/
- protected def genNamesParser(names: Seq[String]): Parser[String] = {
+ protected def genDataSourceNamesParser(names: Seq[String]): Parser[String] = {
+ names.reverse.map {
+ fn => s"""(?i)`${fn}`|${fn}""".r: Parser[String]
+ }.reduce(_ | _)
+ }
+ protected def genFunctionNamesParser(names: Seq[String]): Parser[String] = {
names.reverse.map {
fn => s"""(?i)${fn}""".r: Parser[String]
}.reduce(_ | _)
@@ -143,9 +148,9 @@ trait BasicParser extends JavaTokenParsers with Serializable {
object Strings {
def AnyString: Parser[String] = """"(?:[^\"]|\")*"""".r | """'(?:[^']|\')*'""".r
def UQuoteTableFieldName: Parser[String] = """`(?:[^`]|[\\][`])*`""".r
- def TableFieldName: Parser[String] = UQuoteTableFieldName | """[a-zA-Z_]\w*""".r
- def DataSourceName: Parser[String] = genNamesParser(dataSourceNames)
- def FunctionName: Parser[String] = genNamesParser(functionNames)
+ def FieldName: Parser[String] = UQuoteTableFieldName | """[a-zA-Z_]\w*""".r
+ def DataSourceName: Parser[String] = genDataSourceNamesParser(dataSourceNames)
+ def FunctionName: Parser[String] = genFunctionNamesParser(functionNames)
def IntegerNumber: Parser[String] = """[+\-]?\d+""".r
def DoubleNumber: Parser[String] = """[+\-]?(\.\d+|\d+\.\d*)""".r
@@ -191,20 +196,20 @@ trait BasicParser extends JavaTokenParsers with Serializable {
DataSourceHeadExpr(_)
} | function ^^ {
OtherHeadExpr(_)
- } | TableFieldName ^^ {
+ } | FieldName ^^ {
FieldNameHeadExpr(_)
} | ALLSL ^^ { _ =>
ALLSelectHeadExpr()
}
def selector: Parser[SelectExpr] = functionSelect | allFieldsSelect | fieldSelect | indexSelect
def allFieldsSelect: Parser[AllFieldsSelectExpr] = DOT ~> ALLSL ^^ { _ => AllFieldsSelectExpr() }
- def fieldSelect: Parser[FieldSelectExpr] = DOT ~> TableFieldName ^^ { FieldSelectExpr(_) }
+ def fieldSelect: Parser[FieldSelectExpr] = DOT ~> FieldName ^^ { FieldSelectExpr(_) }
def indexSelect: Parser[IndexSelectExpr] = LSQBR ~> argument <~ RSQBR ^^ { IndexSelectExpr(_) }
def functionSelect: Parser[FunctionSelectExpr] = DOT ~ FunctionName ~ LBR ~ repsep(argument, COMMA) ~ RBR ^^ {
case _ ~ name ~ _ ~ args ~ _ => FunctionSelectExpr(name, args)
}
- def asAlias: Parser[String] = AS ~> TableFieldName
+ def asAlias: Parser[String] = AS ~> FieldName
/**
* -- math expr --
@@ -310,6 +315,7 @@ trait BasicParser extends JavaTokenParsers with Serializable {
* -- clauses --
* <select-clause> = <expr> [, <expr>]*
* <where-clause> = <where> <expr>
+ * <from-clause> = <from> ("data source name registered")
* <having-clause> = <having> <expr>
* <groupby-clause> = <group> <by> <expr> [ <having-clause> ]?
* <orderby-item> = <expr> [ <DESC> ]?
@@ -318,7 +324,7 @@ trait BasicParser extends JavaTokenParsers with Serializable {
*/
def selectClause: Parser[SelectClause] = opt(SELECT) ~> rep1sep(expression, COMMA) ^^ { SelectClause(_) }
- def fromClause: Parser[FromClause] = FROM ~> TableFieldName ^^ { ds => FromClause(trim(ds)) }
+ def fromClause: Parser[FromClause] = FROM ~> DataSourceName ^^ { ds => FromClause(trim(ds)) }
def whereClause: Parser[WhereClause] = WHERE ~> expression ^^ { WhereClause(_) }
def havingClause: Parser[Expr] = HAVING ~> expression
def groupbyClause: Parser[GroupbyClause] = GROUP ~ BY ~ rep1sep(expression, COMMA) ~ opt(havingClause) ^^ {
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/test/resources/config-test-accuracy-streaming-multids.json
----------------------------------------------------------------------
diff --git a/measure/src/test/resources/config-test-accuracy-streaming-multids.json b/measure/src/test/resources/config-test-accuracy-streaming-multids.json
index 18532b0..2107ff6 100644
--- a/measure/src/test/resources/config-test-accuracy-streaming-multids.json
+++ b/measure/src/test/resources/config-test-accuracy-streaming-multids.json
@@ -51,7 +51,6 @@
"dsl.type": "df-opr",
"name": "${s1}",
"rule": "from_json",
- "persist.type": "cache",
"details": {
"df.name": "${this}"
}
@@ -93,7 +92,6 @@
"dsl.type": "df-opr",
"name": "${t1}",
"rule": "from_json",
- "persist.type": "cache",
"details": {
"df.name": "${this}"
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/test/resources/config-test-accuracy-streaming.json
----------------------------------------------------------------------
diff --git a/measure/src/test/resources/config-test-accuracy-streaming.json b/measure/src/test/resources/config-test-accuracy-streaming.json
index 276f8dd..453f87e 100644
--- a/measure/src/test/resources/config-test-accuracy-streaming.json
+++ b/measure/src/test/resources/config-test-accuracy-streaming.json
@@ -26,7 +26,6 @@
"dsl.type": "df-opr",
"name": "${s1}",
"rule": "from_json",
- "persist.type": "cache",
"details": {
"df.name": "${this}"
}
@@ -68,7 +67,6 @@
"dsl.type": "df-opr",
"name": "${t1}",
"rule": "from_json",
- "persist.type": "cache",
"details": {
"df.name": "${this}"
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/test/scala/org/apache/griffin/measure/data/connector/ConnectorTest.scala
----------------------------------------------------------------------
diff --git a/measure/src/test/scala/org/apache/griffin/measure/data/connector/ConnectorTest.scala b/measure/src/test/scala/org/apache/griffin/measure/data/connector/ConnectorTest.scala
index ead84f7..30cacc4 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/data/connector/ConnectorTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/data/connector/ConnectorTest.scala
@@ -67,5 +67,31 @@ class ConnectorTest extends FunSuite with Matchers with BeforeAndAfter {
val clazz = Class.forName(tp)
ClassTag(clazz)
}
+
+ test ("test sql") {
+ val concreteTableName = "default.demo_src"
+ val partitionsString = " dt=1, hr=2; dt=1 ; ;,hr=4; dt=2 ;;dt=5"
+ val partitions: Array[Array[String]] = partitionsString.split(";").flatMap { s =>
+ val arr = s.trim.split(",").flatMap { t =>
+ t.trim match {
+ case p if (p.nonEmpty) => Some(p)
+ case _ => None
+ }
+ }
+ if (arr.size > 0) Some(arr) else None
+ }
+
+ val tableClause = s"SELECT * FROM ${concreteTableName}"
+ val validPartitions = partitions.filter(_.size > 0)
+ val ret = if (validPartitions.size > 0) {
+ val clauses = validPartitions.map { prtn =>
+ val cls = prtn.mkString(" AND ")
+ s"${tableClause} WHERE ${cls}"
+ }
+ clauses.mkString(" UNION ALL ")
+ } else tableClause
+
+ println(ret)
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/measure/src/test/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParserTest.scala
----------------------------------------------------------------------
diff --git a/measure/src/test/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParserTest.scala b/measure/src/test/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParserTest.scala
index 823554d..a1b9a83 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParserTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/rule/dsl/parser/BasicParserTest.scala
@@ -164,11 +164,13 @@ class BasicParserTest extends FunSuite with Matchers with BeforeAndAfter {
test ("test expression") {
val rule3 = "source.age + 2 * 5 + target.offset"
val result3 = parser.parseAll(parser.expression, rule3)
+ println(result3)
result3.successful should be (true)
result3.get.desc should be ("source.age + 2 * 5 + target.offset")
val rule4 = "(source.age + 2) * (5 + target.offset)"
val result4 = parser.parseAll(parser.expression, rule4)
+ println(result4)
result4.successful should be (true)
result4.get.desc should be ("(source.age + 2) * (5 + target.offset)")
@@ -203,14 +205,15 @@ class BasicParserTest extends FunSuite with Matchers with BeforeAndAfter {
}
test ("select clause") {
- val rule = "select source.user_id, item, source.func()"
+ val rule = "select `source`.user_id, item, `source`.age.func()"
val result = parser.parseAll(parser.selectClause, rule)
+ println(result)
result.successful should be (true)
println(result.get.desc)
}
test ("from clause") {
- val rule = "from source"
+ val rule = "from `source`"
val result = parser.parseAll(parser.fromClause, rule)
result.successful should be (true)
println(result.get.desc)
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/service/src/main/java/org/apache/griffin/core/measure/repo/DataSourceRepo.java
----------------------------------------------------------------------
diff --git a/service/src/main/java/org/apache/griffin/core/measure/repo/DataSourceRepo.java b/service/src/main/java/org/apache/griffin/core/measure/repo/DataSourceRepo.java
deleted file mode 100644
index 5cc897c..0000000
--- a/service/src/main/java/org/apache/griffin/core/measure/repo/DataSourceRepo.java
+++ /dev/null
@@ -1,26 +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.apache.griffin.core.measure.repo;
-
-import org.apache.griffin.core.measure.entity.DataSource;
-import org.springframework.data.repository.CrudRepository;
-
-public interface DataSourceRepo extends CrudRepository<DataSource,Long> {
-}
http://git-wip-us.apache.org/repos/asf/incubator-griffin/blob/abda11ef/service/src/main/java/org/apache/griffin/core/measure/repo/RuleRepo.java
----------------------------------------------------------------------
diff --git a/service/src/main/java/org/apache/griffin/core/measure/repo/RuleRepo.java b/service/src/main/java/org/apache/griffin/core/measure/repo/RuleRepo.java
deleted file mode 100644
index 75adce5..0000000
--- a/service/src/main/java/org/apache/griffin/core/measure/repo/RuleRepo.java
+++ /dev/null
@@ -1,26 +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.apache.griffin.core.measure.repo;
-
-import org.apache.griffin.core.measure.entity.Rule;
-import org.springframework.data.repository.CrudRepository;
-
-public interface RuleRepo extends CrudRepository<Rule,Long> {
-}