[GRIFFIN-347] Updated with master
diff --git a/.gitignore b/.gitignore
index 93610da..8d3873e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -2,6 +2,7 @@
*.iml
.idea/
.DS_Store
+.java-version
target/**
# Mobile Tools for Java (J2ME)
@@ -11,6 +12,7 @@
*.jar
*.war
*.ear
+*.cfg
target
service/src/main/resources/public/
ui/.tmp/
@@ -20,7 +22,7 @@
.settings/
.classpath
bin
-
+**/site-packages/**
# virtual machine crash logs, see http://www.java.com/en/download/help/error_hotspot.xml
hs_err_pid*
# don't need binary
diff --git a/griffin-doc/deploy/measure-build-guide.md b/griffin-doc/deploy/measure-build-guide.md
new file mode 100644
index 0000000..8914bc6
--- /dev/null
+++ b/griffin-doc/deploy/measure-build-guide.md
@@ -0,0 +1,74 @@
+<!--
+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.
+-->
+
+# Apache Griffin Build Guide - Measure Module
+
+Like other modules of Apache Griffin, `measure` module is also built using Maven build tool. Building `measure` module
+requires Maven version 3.5+ and Java 8.
+
+## Version Compatibility
+
+Starting from Apache Griffin 0.7, the `measure` module will be (scala-spark) cross version compatible. Since both Scala
+and Spark are dependencies for Apache Griffin, details of Spark-Scala cross version compatibility is mentioned below,
+
+| | Spark 2.3.x | Spark 2.4.x | Spark 3.0.x |
+| --------- |:-----------:|:-----------:|:-----------:|
+| Scala 2.11| ✓ | ✓ | x |
+| Scala 2.12| x | ✓ | ✓ |
+
+## Building a Distribution
+
+Execute the below commands to build `measure` with desired version of Spark and Scala,
+
+By default, the build is compiled with Scala 2.11 and Apache Spark 2.4.x.
+
+```
+# For measure module with Scala 2.11 and Spark 2.4.x
+mvn clean package
+```
+
+To change Scala or Spark version you can use the commands below,
+
+```
+# For measure module with Scala 2.12 and Spark 2.4.x
+mvn clean package -Dscala-2.12
+```
+
+```
+# For measure module with Scala 2.11 and Spark 2.3.x
+mvn clean package -Dspark-2.3
+```
+
+```
+# For measure module with Scala 2.12 and Spark 3.0.x
+mvn clean package -Dscala-2.12 -Dspark-3.0
+```
+
+Note:
+ - Using `-Dscala-2.12` and `-Dspark-2.3` option together will cause build failure due to missing dependencies as it
+is not cross compiled, see details [here](#version-compatibility)
+ - Using `-Dspark-3.0` option without `-Dscala-2.12` option will cause build failure due to missing dependencies as it
+ is not cross compiled, see details [here](#version-compatibility)
+
+### AVRO Source Support
+
+Starting Spark 2.4.x, AVRO source (`spark-avro` package) was migrated from `com.databricks` group to `org.apache.spark`.
+Additionally, the older dependency does not support scala 2.12.
+
+All builds of `measure` module will contain AVRO source support by default.
diff --git a/griffin-doc/measure/predicates.md b/griffin-doc/measure/predicates.md
index 87f1a0d..f9187b9 100644
--- a/griffin-doc/measure/predicates.md
+++ b/griffin-doc/measure/predicates.md
@@ -17,14 +17,13 @@
under the License.
-->
-#About predicates
+# About predicates
-##Overview
+## Overview
The purpose of predicates is obligate Griffin to check certain conditions before starting SparkSubmitJob.
Depending on these conditions Griffin need to start or not start the measurement.
-##Configure predicates
-
+## Configure predicates
For configuring predicates need add property to measure json:
```
{
@@ -74,10 +73,10 @@
- implement interface **org.apache.griffin.core.job.Predicator**
- have constructor with argument of type **org.apache.griffin.core.job.entity.SegmentPredicate**
-##Deployment custom predicates
+## Deployment custom predicates
For the creating custom predicate you need
1. Build the Griffin service using command
-As a result, two artifacts will be built
+As a result, two artifacts will be built,
- **service-VERSION.jar** - executable Spring-Boot application
- **service-VERSION-lib.jar** - jar, which we can use as a dependency
This step is necessary because we can't use executable Spring-Boot application as a dependency in our plugin.
diff --git a/measure/assembly.xml b/measure/assembly.xml
deleted file mode 100644
index 7890497..0000000
--- a/measure/assembly.xml
+++ /dev/null
@@ -1,64 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<!--
- ~ 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.
- -->
-
-<assembly xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2"
- xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
- xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.2 http://maven.apache.org/xsd/assembly-1.1.2.xsd">
- <id>package</id>
- <formats>
- <format>tar.gz</format>
- </formats>
- <fileSets>
- <fileSet>
- <directory>${project.basedir}/sbin</directory>
- <outputDirectory>/bin</outputDirectory>
- <includes>
- <include>/**</include>
- </includes>
- <lineEnding>unix</lineEnding>
- <fileMode>0777</fileMode>
- <directoryMode>0755</directoryMode>
- </fileSet>
- <fileSet>
- <directory>${project.basedir}/src/main/resources</directory>
- <outputDirectory>/conf</outputDirectory>
- <includes>
- <include>/**</include>
- </includes>
- </fileSet>
- </fileSets>
-
- <dependencySets>
- <dependencySet>
- <useProjectArtifact>true</useProjectArtifact>
- <outputDirectory>/lib</outputDirectory>
- <unpack>false</unpack>
- <scope>provided</scope>
- <excludes>
- <exclude>com.twitter:parquet-hadoop-bundle</exclude>
- <exclude>io.dropwizard.metrics:*</exclude>
- <exclude>org.glassfish.jersey.core:*</exclude>
- <exclude>org.glassfish.jersey.containers:*</exclude>
- <exclude>org.apache.thrift:*</exclude>
- <exclude>org.apache.parquet:*</exclude>
- <exclude>org.apache.hadoop:*</exclude>
- <exclude>org.apache.spark:*</exclude>
- </excludes>
- </dependencySet>
- </dependencySets>
-</assembly>
diff --git a/measure/pom.xml b/measure/pom.xml
index a1d8f8e..3cd8bc6 100644
--- a/measure/pom.xml
+++ b/measure/pom.xml
@@ -24,297 +24,314 @@
<parent>
<groupId>org.apache.griffin</groupId>
<artifactId>griffin</artifactId>
- <version>0.6.0-SNAPSHOT</version>
+ <version>0.7.0-SNAPSHOT</version>
</parent>
<artifactId>measure</artifactId>
<packaging>jar</packaging>
- <name>Apache Griffin :: Measures</name>
- <url>http://maven.apache.org</url>
+ <name>Apache Griffin :: Measure</name>
+ <url>http://griffin.apache.org</url>
<properties>
- <scala.version>2.11.8</scala.version>
- <spark.version>2.2.1</spark.version>
- <scala.binary.version>2.11</scala.binary.version>
- <avro.version>1.7.7</avro.version>
- <jackson.version>2.8.7</jackson.version>
+
+ <!-- Spark -->
+ <spark.major.version>2.4</spark.major.version>
+ <spark.version>${spark.major.version}.4</spark.version>
+ <spark.scope>${provided.scope}</spark.scope>
+ <spark-streaming-kafka.version>${spark.version}</spark-streaming-kafka.version>
+
+ <!-- Code Standardization -->
+ <scalastyle.version>1.0.0</scalastyle.version>
+ <scalafmt.version>1.0.3</scalafmt.version>
+
+ <!-- Data Connectors -->
+ <mysql.java.version>5.1.47</mysql.java.version>
+ <mariadb.version>2.7.0</mariadb.version>
+ <cassandra.connector.version>2.5.1</cassandra.connector.version>
+ <elasticsearch.version>6.4.1</elasticsearch.version>
<scalaj.version>2.3.0</scalaj.version>
<mongo.version>2.1.0</mongo.version>
- <scalatest.version>3.0.0</scalatest.version>
- <slf4j.version>1.7.21</slf4j.version>
- <log4j.version>1.2.16</log4j.version>
<curator.version>2.10.0</curator.version>
- <mockito.version>1.10.19</mockito.version>
- <mysql.java.version>5.1.47</mysql.java.version>
- <cassandra.connector.version>2.4.1</cassandra.connector.version>
- <scalastyle.version>1.0.0</scalastyle.version>
- <scalafmt.parameters>--diff --test</scalafmt.parameters>
- <scalafmt.skip>false</scalafmt.skip>
- <elasticsearch.version>6.4.1</elasticsearch.version>
- <spark.scope>provided</spark.scope>
+
+ <!-- Testing -->
+ <scoverage.plugin.version>1.4.0</scoverage.plugin.version>
+ <mockito.version>3.2.2.0</mockito.version>
+ <scalatest.version>3.2.3</scalatest.version>
</properties>
<dependencies>
- <!--scala-->
+
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
</dependency>
- <!--spark, spark streaming, spark hive-->
+ <!-- Spark Dependencies -->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_${scala.binary.version}</artifactId>
<version>${spark.version}</version>
<scope>${spark.scope}</scope>
</dependency>
- <dependency>
- <groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming_${scala.binary.version}</artifactId>
- <version>${spark.version}</version>
- <scope>${spark.scope}</scope>
- </dependency>
+
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_${scala.binary.version}</artifactId>
<version>${spark.version}</version>
<scope>${spark.scope}</scope>
</dependency>
+
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-streaming_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ <scope>${spark.scope}</scope>
+ </dependency>
+
+ <!-- Data Connectors -->
+
+ <!-- Spark Hive-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-hive_${scala.binary.version}</artifactId>
<version>${spark.version}</version>
<scope>${spark.scope}</scope>
+ </dependency>
+
+ <!-- Spark Kafka-->
+ <!-- TODO
+ Spark Streaming Kafka 08 is not cross compiled with scala 2.12.
+ Thus, this dependency has been temporary un-banned in the enforcer plugin.
+ This will be removed with structured streaming implementation
+ -->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-streaming-kafka-0-8_${scala211.binary.version}</artifactId>
+ <version>${spark-streaming-kafka.version}</version>
<exclusions>
<exclusion>
- <groupId>commons-httpclient</groupId>
- <artifactId>commons-httpclient</artifactId>
+ <groupId>org.scala-lang.modules</groupId>
+ <artifactId>scala-xml_2.11</artifactId>
</exclusion>
<exclusion>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
+ <groupId>org.scala-lang.modules</groupId>
+ <artifactId>scala-parser-combinators_2.11</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-tags_2.11</artifactId>
</exclusion>
</exclusions>
</dependency>
- <dependency>
- <groupId>org.apache.spark</groupId>
- <artifactId>spark-streaming-kafka-0-8_${scala.binary.version}</artifactId>
- <version>${spark.version}</version>
- </dependency>
- <!--jackson-->
- <dependency>
- <groupId>com.fasterxml.jackson.core</groupId>
- <artifactId>jackson-databind</artifactId>
- <version>${jackson.version}</version>
- </dependency>
- <dependency>
- <groupId>com.fasterxml.jackson.module</groupId>
- <artifactId>jackson-module-scala_${scala.binary.version}</artifactId>
- <version>${jackson.version}</version>
- </dependency>
-
- <!--scalaj for http request-->
- <dependency>
- <groupId>org.scalaj</groupId>
- <artifactId>scalaj-http_${scala.binary.version}</artifactId>
- <version>${scalaj.version}</version>
- </dependency>
-
- <!--mongo request-->
+ <!-- MongoDB-->
<dependency>
<groupId>org.mongodb.scala</groupId>
- <artifactId>mongo-scala-driver_2.11</artifactId>
+ <artifactId>mongo-scala-driver_${scala.binary.version}</artifactId>
<version>${mongo.version}</version>
</dependency>
- <!--avro-->
+ <!-- MySql -->
<dependency>
- <groupId>com.databricks</groupId>
- <artifactId>spark-avro_${scala.binary.version}</artifactId>
- <version>4.0.0</version>
+ <groupId>mysql</groupId>
+ <artifactId>mysql-connector-java</artifactId>
+ <version>${mysql.java.version}</version>
</dependency>
- <!--log4j-->
+ <!-- MariaDB -->
<dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- <version>${slf4j.version}</version>
+ <groupId>org.mariadb.jdbc</groupId>
+ <artifactId>mariadb-java-client</artifactId>
+ <version>${mariadb.version}</version>
</dependency>
+ <!-- Spark Cassandra-->
<dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-log4j12</artifactId>
- <version>${slf4j.version}</version>
- </dependency>
- <dependency>
- <groupId>log4j</groupId>
- <artifactId>log4j</artifactId>
- <version>${log4j.version}</version>
+ <groupId>com.datastax.spark</groupId>
+ <artifactId>spark-cassandra-connector_${scala.binary.version}</artifactId>
+ <version>${cassandra.connector.version}</version>
</dependency>
- <!--curator-->
+ <!-- Spark Elasticsearch-->
+ <!-- TODO
+ Elasticsearch Spark has recently added support for scala 2.12 but it is not published in the Maven Central yet.
+ Thus, this dependency has been temporary un-banned in the enforcer plugin.
+ See https://github.com/elastic/elasticsearch-hadoop/pull/1589#issuecomment-776920189
+ -->
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch-spark-20_${scala211.binary.version}</artifactId>
+ <version>${elasticsearch.version}</version>
+ </dependency>
+
+ <!-- Miscellaneous-->
<dependency>
<groupId>org.apache.curator</groupId>
<artifactId>curator-recipes</artifactId>
<version>${curator.version}</version>
</dependency>
- <!--junit-->
+ <dependency>
+ <groupId>org.scalaj</groupId>
+ <artifactId>scalaj-http_${scala.binary.version}</artifactId>
+ <version>${scalaj.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ <version>4.5.9</version>
+ </dependency>
+
+ <!-- Test -->
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
+ <scope>test</scope>
</dependency>
- <!--scala test-->
<dependency>
<groupId>org.scalatest</groupId>
<artifactId>scalatest_${scala.binary.version}</artifactId>
<version>${scalatest.version}</version>
<scope>test</scope>
</dependency>
+
<dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <version>${mockito.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-all</artifactId>
+ <groupId>org.scalatestplus</groupId>
+ <artifactId>mockito-3-4_${scala.binary.version}</artifactId>
<version>${mockito.version}</version>
<scope>test</scope>
</dependency>
<dependency>
- <groupId>mysql</groupId>
- <artifactId>mysql-connector-java</artifactId>
- <version>${mysql.java.version}</version>
- </dependency>
- <dependency>
- <groupId>com.datastax.spark</groupId>
- <artifactId>spark-cassandra-connector_2.11</artifactId>
- <version>${cassandra.connector.version}</version>
- </dependency>
- <dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- <version>4.5.9</version>
- </dependency>
- <dependency>
<groupId>com.h2database</groupId>
<artifactId>h2</artifactId>
<version>1.4.200</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.elasticsearch</groupId>
- <artifactId>elasticsearch-spark-20_2.11</artifactId>
- <version>${elasticsearch.version}</version>
- </dependency>
+
<dependency>
<groupId>org.testcontainers</groupId>
<artifactId>elasticsearch</artifactId>
<version>1.14.3</version>
<scope>test</scope>
</dependency>
+
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<testSourceDirectory>src/test/scala</testSourceDirectory>
<plugins>
+
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
- <version>3.3.1</version>
+ <version>4.3.0</version>
<executions>
<execution>
- <id>compile</id>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
- <phase>compile</phase>
</execution>
</executions>
<configuration>
<scalaVersion>${scala.version}</scalaVersion>
+ <checkMultipleScalaVersions>true</checkMultipleScalaVersions>
+ <failOnMultipleScalaVersions>true</failOnMultipleScalaVersions>
+ <recompileMode>incremental</recompileMode>
<args>
+ <arg>-unchecked</arg>
<arg>-deprecation</arg>
<arg>-feature</arg>
- <arg>-unchecked</arg>
+ <arg>-explaintypes</arg>
</args>
+ <jvmArgs>
+ <jvmArg>-Xms64m</jvmArg>
+ <jvmArg>-Xmx1024m</jvmArg>
+ </jvmArgs>
+ <javacArgs>
+ <javacArg>-source</javacArg>
+ <javacArg>-target</javacArg>
+ <javacArg>${java.version}</javacArg>
+ <javacArg>-Xlint:all,-serial,-path,-try</javacArg>
+ </javacArgs>
</configuration>
</plugin>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- </plugin>
+
<plugin>
<groupId>org.scalatest</groupId>
<artifactId>scalatest-maven-plugin</artifactId>
+ <version>2.0.2</version>
+ <configuration>
+ <testFailureIgnore>false</testFailureIgnore>
+ <reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
+ <junitxml>.</junitxml>
+ <filereports>TestSuiteReport.txt</filereports>
+ <stderr/>
+ <systemProperties>
+ <java.io.tmpdir>${project.build.directory}/tmp</java.io.tmpdir>
+ </systemProperties>
+ </configuration>
+ <executions>
+ <execution>
+ <id>test</id>
+ <goals>
+ <goal>test</goal>
+ </goals>
+ </execution>
+ </executions>
</plugin>
+
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
- <version>3.1.0</version>
- <configuration>
- <createDependencyReducedPom>false</createDependencyReducedPom>
- </configuration>
+ <version>3.2.4</version>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
- <configuration>
- <transformers>
- <transformer
- implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
- <mainClass>org.apache.griffin.measure.Application</mainClass>
- </transformer>
- </transformers>
- <relocations>
- <relocation>
- <pattern>org.apache.http</pattern>
- <shadedPattern>griffin.org.apache.http</shadedPattern>
- </relocation>
- </relocations>
- <filters>
- <filter>
- <artifact>*:*</artifact>
- <excludes>
- <exclude>META-INF/maven/**</exclude>
- </excludes>
- </filter>
- </filters>
- </configuration>
</execution>
</executions>
+ <configuration>
+ <createDependencyReducedPom>false</createDependencyReducedPom>
+ <minimizeJar>true</minimizeJar>
+ <relocations>
+ <relocation>
+ <pattern>org.apache.http</pattern>
+ <shadedPattern>griffin.org.apache.http</shadedPattern>
+ </relocation>
+ </relocations>
+ </configuration>
</plugin>
+
<plugin>
<groupId>org.antipathy</groupId>
<artifactId>mvn-scalafmt_${scala.binary.version}</artifactId>
- <version>0.12_1.5.1</version>
+ <version>${scalafmt.version}</version>
<configuration>
- <parameters>${scalafmt.parameters}</parameters>
- <skip>${scalafmt.skip}</skip>
- <skipSources>${scalafmt.skip}</skipSources>
- <skipTestSources>${scalafmt.skip}</skipTestSources>
<configLocation>${project.parent.basedir}/.scalafmt.conf</configLocation>
+ <skipTestSources>false</skipTestSources>
+ <skipSources>false</skipSources>
+ <respectVersion>false</respectVersion>
+ <validateOnly>false</validateOnly>
</configuration>
<executions>
<execution>
- <phase>validate</phase>
+ <phase>compile</phase>
<goals>
<goal>format</goal>
</goals>
</execution>
</executions>
</plugin>
+
<plugin>
<groupId>org.scalastyle</groupId>
<artifactId>scalastyle-maven-plugin</artifactId>
@@ -333,21 +350,81 @@
</configuration>
<executions>
<execution>
- <phase>validate</phase>
+ <phase>compile</phase>
<goals>
<goal>check</goal>
</goals>
</execution>
</executions>
</plugin>
+
+ <plugin>
+ <groupId>org.jacoco</groupId>
+ <artifactId>jacoco-maven-plugin</artifactId>
+ <version>0.8.2</version>
+ <executions>
+ <execution>
+ <id>default-prepare-agent</id>
+ <goals>
+ <goal>prepare-agent</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>report</id>
+ <phase>test</phase>
+ <goals>
+ <goal>report</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+
+ <plugin>
+ <groupId>org.scoverage</groupId>
+ <artifactId>scoverage-maven-plugin</artifactId>
+ <version>${scoverage.plugin.version}</version>
+ <configuration>
+ <scalaVersion>${scala.version}</scalaVersion>
+ <aggregate>true</aggregate>
+ <highlighting>true</highlighting>
+ </configuration>
+ </plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-jar-plugin</artifactId>
+ <version>3.2.0</version>
+ <executions>
+ <execution>
+ <phase>package</phase>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ <configuration>
+ <finalName>${project.artifactId}-${project.version}</finalName>
+ <classifier>spark-${spark.major.version}_scala-${scala.binary.version}</classifier>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-assembly-plugin</artifactId>
+ <version>3.3.0</version>
<configuration>
- <descriptors>
- <descriptor>assembly.xml</descriptor>
- </descriptors>
- <finalName>${artifactId}-${project.version}</finalName>
+ <tarLongFileMode>posix</tarLongFileMode>
+ <finalName>
+ ${project.artifactId}-${project.version}-spark-${spark.major.version}_scala-${scala.binary.version}
+ </finalName>
+ <archive>
+ <manifest>
+ <mainClass>org.apache.griffin.measure.Application</mainClass>
+ </manifest>
+ </archive>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
</configuration>
<executions>
<execution>
@@ -359,6 +436,172 @@
</execution>
</executions>
</plugin>
+
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <version>3.0.0-M2</version>
+ <executions>
+ <execution>
+ <id>enforce-versions</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <requireJavaVersion>
+ <version>${java.version}</version>
+ </requireJavaVersion>
+ <bannedDependencies>
+ <excludes>
+ <exclude>*:*_2.10</exclude>
+ <exclude>*:*_2.12</exclude>
+ <exclude>*:*_2.13</exclude>
+ </excludes>
+ <searchTransitive>true</searchTransitive>
+ </bannedDependencies>
+ </rules>
+ </configuration>
+ </execution>
+ <execution>
+ <id>enforce-no-duplicate-dependencies</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <banDuplicatePomDependencyVersions/>
+ </rules>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
</plugins>
</build>
+
+ <profiles>
+ <!-- Scala 2.12 -->
+ <profile>
+ <id>scala-2.12</id>
+ <activation>
+ <property>
+ <name>scala-2.12</name>
+ </property>
+ </activation>
+
+ <properties>
+ <scala.binary.version>2.12</scala.binary.version>
+ <scala.version>${scala.binary.version}.12</scala.version>
+ </properties>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-enforcer-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>enforce-versions</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <requireJavaVersion>
+ <version>${java.version}</version>
+ </requireJavaVersion>
+ <bannedDependencies>
+ <!-- TODO Temporary Fix: These will be removed in a future iteration. -->
+ <includes>
+ <include>org.elasticsearch:elasticsearch-spark-20_2.11:*</include>
+ <include>org.apache.spark:spark-streaming-kafka-0-8_2.11:*</include>
+ <include>org.apache.kafka:kafka_2.11:*</include>
+ </includes>
+ <excludes>
+ <exclude>*:*_2.10</exclude>
+ <exclude>*:*_2.11</exclude>
+ <exclude>*:*_2.13</exclude>
+ </excludes>
+ <searchTransitive>true</searchTransitive>
+ </bannedDependencies>
+ </rules>
+ </configuration>
+ </execution>
+ <execution>
+ <id>enforce-no-duplicate-dependencies</id>
+ <goals>
+ <goal>enforce</goal>
+ </goals>
+ <configuration>
+ <rules>
+ <banDuplicatePomDependencyVersions/>
+ </rules>
+ </configuration>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+ </profile>
+
+ <!-- Spark 2.3.x -->
+ <profile>
+ <id>spark-2.3</id>
+ <activation>
+ <property>
+ <name>spark-2.3</name>
+ </property>
+ </activation>
+
+ <properties>
+ <spark.major.version>2.3</spark.major.version>
+ <spark.version>${spark.major.version}.4</spark.version>
+ </properties>
+
+ <dependencies>
+ <!-- Old Spark Avro-->
+ <dependency>
+ <groupId>com.databricks</groupId>
+ <artifactId>spark-avro_${scala211.binary.version}</artifactId>
+ <version>4.0.0</version>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ <!-- Spark 3.0.x -->
+ <profile>
+ <id>spark-3.0</id>
+ <activation>
+ <property>
+ <name>spark-3.0</name>
+ </property>
+ </activation>
+
+ <properties>
+ <spark.major.version>3.0</spark.major.version>
+ <spark.version>${spark.major.version}.2</spark.version>
+ <spark-streaming-kafka.version>2.4.7</spark-streaming-kafka.version>
+ </properties>
+ </profile>
+
+ <!-- Spark 2.4+ Avro -->
+ <profile>
+ <id>spark-avro-2.4</id>
+ <activation>
+ <property>
+ <name>!spark-2.3</name>
+ </property>
+ </activation>
+
+ <dependencies>
+ <!-- New Spark Avro-->
+ <dependency>
+ <groupId>org.apache.spark</groupId>
+ <artifactId>spark-avro_${scala.binary.version}</artifactId>
+ <version>${spark.version}</version>
+ </dependency>
+ </dependencies>
+ </profile>
+
+ </profiles>
</project>
diff --git a/measure/src/main/resources/env-batch.json b/measure/src/main/resources/env-batch.json
index bbec4e5..fd9261d 100644
--- a/measure/src/main/resources/env-batch.json
+++ b/measure/src/main/resources/env-batch.json
@@ -7,14 +7,14 @@
},
"sinks": [
{
- "name": "consoleSink",
+ "name": "console",
"type": "CONSOLE",
"config": {
"max.log.lines": 10
}
},
{
- "name": "hdfsSink",
+ "name": "hdfs",
"type": "HDFS",
"config": {
"path": "hdfs://localhost/griffin/batch/persist",
@@ -23,7 +23,7 @@
}
},
{
- "name": "elasticSink",
+ "name": "elasticsearch",
"type": "ELASTICSEARCH",
"config": {
"method": "post",
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamHttpReader.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamHttpReader.scala
new file mode 100644
index 0000000..e7c19a8
--- /dev/null
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamHttpReader.scala
@@ -0,0 +1,42 @@
+/*
+ * 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.measure.configuration.dqdefinition.reader
+
+import scala.reflect.ClassTag
+import scala.util.Try
+
+import org.apache.griffin.measure.configuration.dqdefinition.Param
+import org.apache.griffin.measure.utils.{HttpUtil, JsonUtil}
+
+/**
+ * read params by http url
+ *
+ * @param httpUrl
+ */
+case class ParamHttpReader(httpUrl: String) extends ParamReader {
+
+ def readConfig[T <: Param](implicit m: ClassTag[T]): Try[T] = {
+ Try {
+ val params = Map[String, Object]()
+ val headers = Map[String, Object](("Content-Type", "application/json"))
+ val jsonString = HttpUtil.doHttpRequest(httpUrl, "get", params, headers, null)._2
+ val param = JsonUtil.fromJson[T](jsonString)
+ validate(param)
+ }
+ }
+}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala
index 5067a9d..f4bfd28 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamReaderFactory.scala
@@ -23,6 +23,7 @@
val json = "json"
val file = "file"
+ val httpRegex = "^http[s]?://.*"
/**
* parse string content to get param reader
@@ -30,9 +31,13 @@
* @return
*/
def getParamReader(pathOrJson: String): ParamReader = {
- val strType = paramStrType(pathOrJson)
- if (json.equals(strType)) ParamJsonReader(pathOrJson)
- else ParamFileReader(pathOrJson)
+ if (pathOrJson.matches(httpRegex)) {
+ ParamHttpReader(pathOrJson)
+ } else {
+ val strType = paramStrType(pathOrJson)
+ if (json.equals(strType)) ParamJsonReader(pathOrJson)
+ else ParamFileReader(pathOrJson)
+ }
}
private def paramStrType(str: String): String = {
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/cache/StreamingCacheClient.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/cache/StreamingCacheClient.scala
index 04775c7..643ff5e 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/cache/StreamingCacheClient.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/cache/StreamingCacheClient.scala
@@ -184,14 +184,15 @@
}
// new cache data
- val newDfOpt = try {
- val dfr = sparkSession.read
- readDataFrameOpt(dfr, newFilePath).map(_.filter(filterStr))
- } catch {
- case e: Throwable =>
- warn(s"read data source cache warn: ${e.getMessage}")
- None
- }
+ val newDfOpt =
+ try {
+ val dfr = sparkSession.read
+ readDataFrameOpt(dfr, newFilePath).map(_.filter(filterStr))
+ } catch {
+ case e: Throwable =>
+ warn(s"read data source cache warn: ${e.getMessage}")
+ None
+ }
// old cache data
val oldCacheIndexOpt = if (updatable) readOldCacheIndex() else None
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
index 39827cf..0b08dd4 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactory.scala
@@ -32,8 +32,15 @@
object DataConnectorFactory extends Loggable {
- @deprecated val AvroRegex: Regex = """^(?i)avro$""".r
- @deprecated val TextDirRegex: Regex = """^(?i)text-dir$""".r
+ @deprecated(
+ s"This class is deprecated. Use '${classOf[FileBasedDataConnector].getCanonicalName}' with correct format.",
+ "0.6.0")
+ val AvroRegex: Regex = """^(?i)avro$""".r
+
+ @deprecated(
+ s"This class is deprecated. Use '${classOf[FileBasedDataConnector].getCanonicalName}' with correct format.",
+ "0.6.0")
+ val TextDirRegex: Regex = """^(?i)text-dir$""".r
val HiveRegex: Regex = """^(?i)hive$""".r
val FileRegex: Regex = """^(?i)file$""".r
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/CassandraDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/CassandraDataConnector.scala
index d135f3b..2fd2912 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/CassandraDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/CassandraDataConnector.scala
@@ -52,33 +52,34 @@
override def data(ms: Long): (Option[DataFrame], TimeRange) = {
- val dfOpt = try {
- sparkSession.conf.set("spark.cassandra.connection.host", host)
- sparkSession.conf.set("spark.cassandra.connection.port", port)
- sparkSession.conf.set("spark.cassandra.auth.username", user)
- sparkSession.conf.set("spark.cassandra.auth.password", password)
+ val dfOpt =
+ try {
+ sparkSession.conf.set("spark.cassandra.connection.host", host)
+ sparkSession.conf.set("spark.cassandra.connection.port", port)
+ sparkSession.conf.set("spark.cassandra.auth.username", user)
+ sparkSession.conf.set("spark.cassandra.auth.password", password)
- val tableDef: DataFrameReader = sparkSession.read
- .format("org.apache.spark.sql.cassandra")
- .options(Map("table" -> tableName, "keyspace" -> database))
+ val tableDef: DataFrameReader = sparkSession.read
+ .format("org.apache.spark.sql.cassandra")
+ .options(Map("table" -> tableName, "keyspace" -> database))
- val dataWh: String = dataWhere()
+ val dataWh: String = dataWhere()
- var data: DataFrame = null
- if (wheres.length > 0) {
- data = tableDef.load().where(dataWh)
- } else {
- data = tableDef.load()
+ var data: DataFrame = null
+ if (wheres.length > 0) {
+ data = tableDef.load().where(dataWh)
+ } else {
+ data = tableDef.load()
+ }
+
+ val dfOpt = Some(data)
+ val preDfOpt = preProcess(dfOpt, ms)
+ preDfOpt
+ } catch {
+ case e: Throwable =>
+ error(s"load cassandra table $database.$TableName fails: ${e.getMessage}", e)
+ None
}
-
- val dfOpt = Some(data)
- val preDfOpt = preProcess(dfOpt, ms)
- preDfOpt
- } catch {
- case e: Throwable =>
- error(s"load cassandra table $database.$TableName fails: ${e.getMessage}", e)
- None
- }
val tmsts = readTmst(ms)
(dfOpt, TimeRange(ms, tmsts))
}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala
index dfdacc8..597695e 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchGriffinDataConnector.scala
@@ -77,23 +77,24 @@
def dataBySql(ms: Long): (Option[DataFrame], TimeRange) = {
val path: String = s"/_sql?format=csv"
info(s"ElasticSearchGriffinDataConnector data : sql: $sql")
- val dfOpt = try {
- val answer = httpPost(path, sql)
- if (answer._1) {
- import sparkSession.implicits._
- val rdd: RDD[String] = sparkSession.sparkContext.parallelize(answer._2.lines.toList)
- val reader: DataFrameReader = sparkSession.read
- reader.option("header", true).option("inferSchema", true)
- val df: DataFrame = reader.csv(rdd.toDS())
- val dfOpt = Some(df)
- val preDfOpt = preProcess(dfOpt, ms)
- preDfOpt
- } else None
- } catch {
- case e: Throwable =>
- error(s"load ES by sql $host:$port $sql fails: ${e.getMessage}", e)
- None
- }
+ val dfOpt =
+ try {
+ val answer = httpPost(path, sql)
+ if (answer._1) {
+ import sparkSession.implicits._
+ val rdd: RDD[String] = sparkSession.sparkContext.parallelize(answer._2.lines.toList)
+ val reader: DataFrameReader = sparkSession.read
+ reader.option("header", true).option("inferSchema", true)
+ val df: DataFrame = reader.csv(rdd.toDS())
+ val dfOpt = Some(df)
+ val preDfOpt = preProcess(dfOpt, ms)
+ preDfOpt
+ } else None
+ } catch {
+ case e: Throwable =>
+ error(s"load ES by sql $host:$port $sql fails: ${e.getMessage}", e)
+ None
+ }
val tmsts = readTmst(ms)
(dfOpt, TimeRange(ms, tmsts))
}
@@ -102,44 +103,44 @@
val path: String = s"/$index/$dataType/_search?sort=tmst:desc&q=name:$metricName&size=$size"
info(s"ElasticSearchGriffinDataConnector data : host: $host port: $port path:$path")
- val dfOpt = try {
- val answer = httpGet(path)
- val data = ArrayBuffer[Map[String, Number]]()
+ val dfOpt =
+ try {
+ val answer = httpGet(path)
+ val data = ArrayBuffer[Map[String, Number]]()
- if (answer._1) {
- val arrayAnswers: util.Iterator[JsonNode] =
- parseString(answer._2).get("hits").get("hits").elements()
+ if (answer._1) {
+ val arrayAnswers: util.Iterator[JsonNode] =
+ parseString(answer._2).get("hits").get("hits").elements()
- while (arrayAnswers.hasNext) {
- val answer = arrayAnswers.next()
- val values = answer.get("_source").get("value")
- val fields: util.Iterator[util.Map.Entry[String, JsonNode]] = values.fields()
- val fieldsMap = mutable.Map[String, Number]()
- while (fields.hasNext) {
- val fld: util.Map.Entry[String, JsonNode] = fields.next()
- fieldsMap.put(fld.getKey, fld.getValue.numberValue())
+ while (arrayAnswers.hasNext) {
+ val answer = arrayAnswers.next()
+ val values = answer.get("_source").get("value")
+ val fields: util.Iterator[util.Map.Entry[String, JsonNode]] = values.fields()
+ val fieldsMap = mutable.Map[String, Number]()
+ while (fields.hasNext) {
+ val fld: util.Map.Entry[String, JsonNode] = fields.next()
+ fieldsMap.put(fld.getKey, fld.getValue.numberValue())
+ }
+ data += fieldsMap.toMap
}
- data += fieldsMap.toMap
}
+ val rdd1: RDD[Map[String, Number]] = sparkSession.sparkContext.parallelize(data)
+ val columns: Array[String] = fields.toArray
+ val defaultNumber: Number = 0.0
+ val rdd: RDD[Row] = rdd1
+ .map { x: Map[String, Number] =>
+ Row(columns.map(c => x.getOrElse(c, defaultNumber).doubleValue()): _*)
+ }
+ val schema = dfSchema(columns.toList)
+ val df: DataFrame = sparkSession.createDataFrame(rdd, schema).limit(size)
+ val dfOpt = Some(df)
+ val preDfOpt = preProcess(dfOpt, ms)
+ preDfOpt
+ } catch {
+ case e: Throwable =>
+ error(s"load ES table $host:$port $index/$dataType fails: ${e.getMessage}", e)
+ None
}
- val rdd1: RDD[Map[String, Number]] = sparkSession.sparkContext.parallelize(data)
- val columns: Array[String] = fields.toArray
- val defaultNumber: Number = 0.0
- val rdd: RDD[Row] = rdd1
- .map { x: Map[String, Number] =>
- Row(columns.map(c => x.getOrElse(c, defaultNumber).doubleValue()): _*)
- }
- val schema = dfSchema(columns.toList)
- val df: DataFrame = sparkSession.createDataFrame(rdd, schema).limit(size)
- df.show(20)
- val dfOpt = Some(df)
- val preDfOpt = preProcess(dfOpt, ms)
- preDfOpt
- } catch {
- case e: Throwable =>
- error(s"load ES table $host:$port $index/$dataType fails: ${e.getMessage}", e)
- None
- }
val tmsts = readTmst(ms)
(dfOpt, TimeRange(ms, tmsts))
}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnector.scala
index 86049bd..2e4f482 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnector.scala
@@ -79,7 +79,14 @@
SupportedFormats.contains(format),
s"Invalid format '$format' specified. Must be one of ${SupportedFormats.mkString("['", "', '", "']")}")
- if (format == "csv") validateCSVOptions()
+ if (format.equalsIgnoreCase("avro") && sparkSession.version < "2.3.0") {
+ format = "com.databricks.spark.avro"
+ }
+
+ if (format == "csv") {
+ validateCSVOptions()
+ }
+
if (format == "tsv") {
format = "csv"
options.getOrElseUpdate(Delimiter, TabDelimiter)
@@ -169,7 +176,8 @@
private val TabDelimiter: String = "\t"
private val DefaultFormat: String = SQLConf.DEFAULT_DATA_SOURCE_NAME.defaultValueString
- private val SupportedFormats: Seq[String] = Seq("parquet", "orc", "avro", "text", "csv", "tsv")
+ private val SupportedFormats: Seq[String] =
+ Seq("parquet", "orc", "avro", "text", "csv", "tsv", "com.databricks.spark.avro")
/**
* Validates the existence of paths in a given sequence.
@@ -189,7 +197,7 @@
else throw new IllegalArgumentException(msg)
false
- })
+ })
assert(validPaths.nonEmpty, "No paths were given for the data source.")
validPaths
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnector.scala
index 9fc21d7..2f77db3 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnector.scala
@@ -69,27 +69,28 @@
assert(isJDBCDriverLoaded(driver), s"JDBC driver $driver not present in classpath")
override def data(ms: Long): (Option[DataFrame], TimeRange) = {
- val dfOpt = try {
- val dtSql = createSqlStmt()
- val prop = new java.util.Properties
- prop.setProperty("user", user)
- prop.setProperty("password", password)
- prop.setProperty("driver", driver)
- val dfOpt = Try(sparkSession.read.jdbc(url, s"($dtSql) as t", prop))
+ val dfOpt =
+ try {
+ val dtSql = createSqlStmt()
+ val prop = new java.util.Properties
+ prop.setProperty("user", user)
+ prop.setProperty("password", password)
+ prop.setProperty("driver", driver)
+ val dfOpt = Try(sparkSession.read.jdbc(url, s"($dtSql) as t", prop))
- dfOpt match {
- case Success(_) =>
- case Failure(exception) =>
- griffinLogger.error("Error occurred while reading data set.", exception)
+ dfOpt match {
+ case Success(_) =>
+ case Failure(exception) =>
+ griffinLogger.error("Error occurred while reading data set.", exception)
+ }
+
+ val preDfOpt = preProcess(dfOpt.toOption, ms)
+ preDfOpt
+ } catch {
+ case e: Throwable =>
+ error(s"loading table $fullTableName fails: ${e.getMessage}", e)
+ None
}
-
- val preDfOpt = preProcess(dfOpt.toOption, ms)
- preDfOpt
- } catch {
- case e: Throwable =>
- error(s"loading table $fullTableName fails: ${e.getMessage}", e)
- None
- }
val tmsts = readTmst(ms)
(dfOpt, TimeRange(ms, tmsts))
}
@@ -99,7 +100,7 @@
*/
private def createSqlStmt(): String = {
val tableClause = s"SELECT * FROM $fullTableName"
- if (whereString.length > 0) {
+ if (whereString.nonEmpty) {
s"$tableClause WHERE $whereString"
} else tableClause
}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/MySqlDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/MySqlDataConnector.scala
index feacfc9..26321fc 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/MySqlDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/batch/MySqlDataConnector.scala
@@ -52,21 +52,22 @@
override def data(ms: Long): (Option[DataFrame], TimeRange) = {
- val dfOpt = try {
- val dtSql = dataSql()
- val prop = new java.util.Properties
- prop.setProperty("user", user)
- prop.setProperty("password", password)
- prop.setProperty("driver", driver)
- val df: DataFrame = sparkSession.read.jdbc(url, s"($dtSql) as t", prop)
- val dfOpt = Some(df)
- val preDfOpt = preProcess(dfOpt, ms)
- preDfOpt
- } catch {
- case e: Throwable =>
- error(s"load mysql table $fullTableName fails: ${e.getMessage}", e)
- None
- }
+ val dfOpt =
+ try {
+ val dtSql = dataSql()
+ val prop = new java.util.Properties
+ prop.setProperty("user", user)
+ prop.setProperty("password", password)
+ prop.setProperty("driver", driver)
+ val df: DataFrame = sparkSession.read.jdbc(url, s"($dtSql) as t", prop)
+ val dfOpt = Some(df)
+ val preDfOpt = preProcess(dfOpt, ms)
+ preDfOpt
+ } catch {
+ case e: Throwable =>
+ error(s"load mysql table $fullTableName fails: ${e.getMessage}", e)
+ None
+ }
val tmsts = readTmst(ms)
(dfOpt, TimeRange(ms, tmsts))
}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingDataConnector.scala b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingDataConnector.scala
index 0c501dc..b0be557 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingDataConnector.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/datasource/connector/streaming/KafkaStreamingDataConnector.scala
@@ -51,23 +51,24 @@
}
ds.foreachRDD((rdd, time) => {
val ms = time.milliseconds
- val saveDfOpt = try {
- // coalesce partition number
- val prlCount = rdd.sparkContext.defaultParallelism
- val ptnCount = rdd.getNumPartitions
- val repartitionedRdd = if (prlCount < ptnCount) {
- rdd.coalesce(prlCount)
- } else rdd
+ val saveDfOpt =
+ try {
+ // coalesce partition number
+ val prlCount = rdd.sparkContext.defaultParallelism
+ val ptnCount = rdd.getNumPartitions
+ val repartitionedRdd = if (prlCount < ptnCount) {
+ rdd.coalesce(prlCount)
+ } else rdd
- val dfOpt = transform(repartitionedRdd)
+ val dfOpt = transform(repartitionedRdd)
- // pre-process
- preProcess(dfOpt, ms)
- } catch {
- case e: Throwable =>
- error(s"streaming data connector error: ${e.getMessage}")
- None
- }
+ // pre-process
+ preProcess(dfOpt, ms)
+ } catch {
+ case e: Throwable =>
+ error(s"streaming data connector error: ${e.getMessage}")
+ None
+ }
// save data frame
streamingCacheClientOpt.foreach(_.saveData(saveDfOpt, ms))
diff --git a/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala b/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
index d8cce41..3134b44 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/sink/ElasticSearchSink.scala
@@ -63,7 +63,8 @@
def func(): (Long, Future[Boolean]) = {
import scala.concurrent.ExecutionContext.Implicits.global
- (timeStamp, Future(HttpUtil.doHttpRequest(api, method, params, header, data)))
+ val code = HttpUtil.doHttpRequest(api, method, params, header, data)._1
+ (timeStamp, Future(code >= 200 && code < 300))
}
if (block) SinkTaskRunner.addBlockTask(func _, retry, connectionTimeout)
else SinkTaskRunner.addNonBlockTask(func _, retry)
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/analyzer/BasicAnalyzer.scala b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/analyzer/BasicAnalyzer.scala
index 743bba9..9aa27f7 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/analyzer/BasicAnalyzer.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/builder/dsl/transform/analyzer/BasicAnalyzer.scala
@@ -43,7 +43,7 @@
v :+ se
case _ => v
}
- }
+ }
val combSelectionExprs: (Seq[SelectionExpr], Seq[SelectionExpr]) => Seq[SelectionExpr] =
(a: Seq[SelectionExpr], b: Seq[SelectionExpr]) => a ++ b
diff --git a/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala b/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
index 4f4bf99..980bded 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/step/write/MetricFlushStep.scala
@@ -33,20 +33,21 @@
def execute(context: DQContext): Try[Boolean] = Try {
context.metricWrapper.flush.foldLeft(true) { (ret, pair) =>
val (t, metric) = pair
- val pr = try {
- context.getSinks(t).foreach { sink =>
- try {
- sink.sinkMetrics(metric)
- } catch {
- case e: Throwable => error(s"sink metrics error: ${e.getMessage}", e)
+ val pr =
+ try {
+ context.getSinks(t).foreach { sink =>
+ try {
+ sink.sinkMetrics(metric)
+ } catch {
+ case e: Throwable => error(s"sink metrics error: ${e.getMessage}", e)
+ }
}
+ true
+ } catch {
+ case e: Throwable =>
+ error(s"flush metrics error: ${e.getMessage}", e)
+ false
}
- true
- } catch {
- case e: Throwable =>
- error(s"flush metrics error: ${e.getMessage}", e)
- false
- }
ret && pr
}
}
diff --git a/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala b/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
index b7f50e8..680f7a8 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/utils/FSUtil.scala
@@ -38,13 +38,14 @@
fsMap.get(uri.getScheme) match {
case Some(fs) => fs
case _ =>
- val fs = try {
- FileSystem.get(uri, getConfiguration)
- } catch {
- case e: Throwable =>
- error(s"get file system error: ${e.getMessage}", e)
- throw e
- }
+ val fs =
+ try {
+ FileSystem.get(uri, getConfiguration)
+ } catch {
+ case e: Throwable =>
+ error(s"get file system error: ${e.getMessage}", e)
+ throw e
+ }
fsMap += (uri.getScheme -> fs)
fs
}
@@ -60,11 +61,12 @@
}
private def getUriOpt(path: String): Option[URI] = {
- val uriOpt = try {
- Some(new URI(path))
- } catch {
- case _: Throwable => None
- }
+ val uriOpt =
+ try {
+ Some(new URI(path))
+ } catch {
+ case _: Throwable => None
+ }
uriOpt.flatMap { uri =>
if (uri.getScheme == null) {
try {
diff --git a/measure/src/main/scala/org/apache/griffin/measure/utils/HttpUtil.scala b/measure/src/main/scala/org/apache/griffin/measure/utils/HttpUtil.scala
index 66648d0..652bbf7 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/utils/HttpUtil.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/utils/HttpUtil.scala
@@ -19,10 +19,10 @@
import scala.util.matching.Regex
-import org.apache.http.client.methods.{HttpGet, HttpPost}
+import org.apache.http.client.methods.{HttpDelete, HttpGet, HttpPost, HttpPut}
+import org.apache.http.client.utils.URIBuilder
import org.apache.http.entity.{ContentType, StringEntity}
-import org.apache.http.impl.client.HttpClientBuilder
-import scalaj.http._
+import org.apache.http.impl.client.{BasicResponseHandler, HttpClientBuilder}
object HttpUtil {
@@ -31,65 +31,34 @@
val PUT_REGEX: Regex = """^(?i)put$""".r
val DELETE_REGEX: Regex = """^(?i)delete$""".r
- def postData(
- url: String,
- params: Map[String, Object],
- headers: Map[String, Object],
- data: String): Boolean = {
- val response = Http(url)
- .params(convertObjMap2StrMap(params))
- .headers(convertObjMap2StrMap(headers))
- .postData(data)
- .asString
-
- response.isSuccess
- }
-
def doHttpRequest(
url: String,
method: String,
params: Map[String, Object],
headers: Map[String, Object],
- data: String): Boolean = {
+ data: String): (Integer, String) = {
val client = HttpClientBuilder.create.build
- method match {
+ val uriBuilder = new URIBuilder(url)
+ convertObjMap2StrMap(params) foreach (param => uriBuilder.setParameter(param._1, param._2))
+ val handler = new BasicResponseHandler()
+ val request = method match {
case POST_REGEX() =>
- val post = new HttpPost(url)
- convertObjMap2StrMap(headers) foreach (header => post.addHeader(header._1, header._2))
+ val post = new HttpPost(uriBuilder.build())
post.setEntity(new StringEntity(data, ContentType.APPLICATION_JSON))
-
- // send the post request
- val response = client.execute(post)
- val code = response.getStatusLine.getStatusCode
- code >= 200 && code < 300
+ post
case PUT_REGEX() =>
- val get = new HttpGet(url)
- convertObjMap2StrMap(headers) foreach (header => get.addHeader(header._1, header._2))
- val response = client.execute(get)
- val code = response.getStatusLine.getStatusCode
- code >= 200 && code < 300
- case _ => false
+ val put = new HttpPut(uriBuilder.build())
+ put.setEntity(new StringEntity(data, ContentType.APPLICATION_JSON))
+ put
+ case GET_REGEX() =>
+ new HttpGet(uriBuilder.build())
+ case DELETE_REGEX() =>
+ new HttpDelete(uriBuilder.build())
+ case _ => throw new UnsupportedOperationException("Unsupported http method error!")
}
- }
-
- def httpRequest(
- url: String,
- method: String,
- params: Map[String, Object],
- headers: Map[String, Object],
- data: String): Boolean = {
- val httpReq = Http(url)
- .params(convertObjMap2StrMap(params))
- .headers(convertObjMap2StrMap(headers))
- method match {
- case POST_REGEX() =>
- val res = httpReq.postData(data).asString
- res.isSuccess
- case PUT_REGEX() =>
- val res = httpReq.put(data).asString
- res.isSuccess
- case _ => false
- }
+ convertObjMap2StrMap(headers) foreach (header => request.addHeader(header._1, header._2))
+ val response = client.execute(request)
+ (response.getStatusLine.getStatusCode, handler.handleResponse(response).trim)
}
private def convertObjMap2StrMap(map: Map[String, Object]): Map[String, String] = {
diff --git a/measure/src/main/scala/org/apache/griffin/measure/utils/ThreadUtils.scala b/measure/src/main/scala/org/apache/griffin/measure/utils/ThreadUtils.scala
index 52c5f38..bc1e292 100644
--- a/measure/src/main/scala/org/apache/griffin/measure/utils/ThreadUtils.scala
+++ b/measure/src/main/scala/org/apache/griffin/measure/utils/ThreadUtils.scala
@@ -21,10 +21,6 @@
import scala.concurrent.{Awaitable, ExecutionContext, ExecutionContextExecutor}
import scala.concurrent.duration.Duration
-import scala.concurrent.forkjoin.{
- ForkJoinPool => SForkJoinPool,
- ForkJoinWorkerThread => SForkJoinWorkerThread
-}
import scala.util.control.NonFatal
import com.google.common.util.concurrent.{MoreExecutors, ThreadFactoryBuilder}
@@ -172,15 +168,15 @@
/**
* Construct a new Scala ForkJoinPool with a specified max parallelism and name prefix.
*/
- def newForkJoinPool(prefix: String, maxThreadNumber: Int): SForkJoinPool = {
+ def newForkJoinPool(prefix: String, maxThreadNumber: Int): ForkJoinPool = {
// Custom factory to set thread names
- val factory = new SForkJoinPool.ForkJoinWorkerThreadFactory {
- override def newThread(pool: SForkJoinPool): SForkJoinWorkerThread =
- new SForkJoinWorkerThread(pool) {
+ val factory = new ForkJoinPool.ForkJoinWorkerThreadFactory {
+ override def newThread(pool: ForkJoinPool): ForkJoinWorkerThread =
+ new ForkJoinWorkerThread(pool) {
setName(prefix + "-" + super.getName)
}
}
- new SForkJoinPool(
+ new ForkJoinPool(
maxThreadNumber,
factory,
null, // handler
diff --git a/measure/src/test/scala/org/apache/griffin/measure/SparkSuiteBase.scala b/measure/src/test/scala/org/apache/griffin/measure/SparkSuiteBase.scala
index dbed89c..011122f 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/SparkSuiteBase.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/SparkSuiteBase.scala
@@ -22,9 +22,10 @@
import org.apache.commons.io.FileUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SparkSession
-import org.scalatest.{BeforeAndAfterAll, FlatSpec}
+import org.scalatest._
+import org.scalatest.flatspec.AnyFlatSpec
-trait SparkSuiteBase extends FlatSpec with BeforeAndAfterAll {
+trait SparkSuiteBase extends AnyFlatSpec with BeforeAndAfterAll {
@transient var spark: SparkSession = _
@transient var sc: SparkContext = _
diff --git a/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamEnumReaderSpec.scala b/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamEnumReaderSpec.scala
index efaa91f..8f191bc 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamEnumReaderSpec.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamEnumReaderSpec.scala
@@ -17,7 +17,9 @@
package org.apache.griffin.measure.configuration.dqdefinition.reader
-import org.scalatest.{FlatSpec, Matchers}
+import org.scalatest._
+import flatspec.AnyFlatSpec
+import matchers.should._
import org.apache.griffin.measure.configuration.dqdefinition.{
DQConfig,
@@ -26,7 +28,7 @@
RuleParam
}
-class ParamEnumReaderSpec extends FlatSpec with Matchers {
+class ParamEnumReaderSpec extends AnyFlatSpec with Matchers {
import org.apache.griffin.measure.configuration.enums.DslType._
"dsltype" should "be parsed to predefined set of values" in {
val validDslSparkSqlValues =
diff --git a/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReaderSpec.scala b/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReaderSpec.scala
index a05df90..94c7e07 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReaderSpec.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamFileReaderSpec.scala
@@ -22,8 +22,10 @@
import org.apache.griffin.measure.configuration.dqdefinition.DQConfig
import org.apache.griffin.measure.configuration.enums.DslType.GriffinDsl
-
-class ParamFileReaderSpec extends FlatSpec with Matchers {
+import org.scalatest._
+import flatspec.AnyFlatSpec
+import matchers.should._
+class ParamFileReaderSpec extends AnyFlatSpec with Matchers {
"params " should "be parsed from a valid file" in {
val reader: ParamReader =
diff --git a/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReaderSpec.scala b/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReaderSpec.scala
index 86d68b5..d8bf125 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReaderSpec.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/configuration/dqdefinition/reader/ParamJsonReaderSpec.scala
@@ -18,14 +18,15 @@
package org.apache.griffin.measure.configuration.dqdefinition.reader
import scala.io.Source
-
-import org.scalatest.{FlatSpec, Matchers}
import scala.util.{Failure, Success}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
+
import org.apache.griffin.measure.configuration.dqdefinition.DQConfig
import org.apache.griffin.measure.configuration.enums.DslType.GriffinDsl
-class ParamJsonReaderSpec extends FlatSpec with Matchers {
+class ParamJsonReaderSpec extends AnyFlatSpec with Matchers {
"params " should "be parsed from a valid file" in {
val bufferedSource =
diff --git a/measure/src/test/scala/org/apache/griffin/measure/context/DataFrameCacheTest.scala b/measure/src/test/scala/org/apache/griffin/measure/context/DataFrameCacheTest.scala
index 898c8fd..6f5717a 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/context/DataFrameCacheTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/context/DataFrameCacheTest.scala
@@ -19,11 +19,12 @@
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.types._
-import org.scalatest._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.SparkSuiteBase
-class DataFrameCacheTest extends FlatSpec with Matchers with SparkSuiteBase {
+class DataFrameCacheTest extends AnyFlatSpec with Matchers with SparkSuiteBase {
def createDataFrame(arr: Seq[Int]): DataFrame = {
val schema = StructType(
diff --git a/measure/src/test/scala/org/apache/griffin/measure/context/MetricWrapperTest.scala b/measure/src/test/scala/org/apache/griffin/measure/context/MetricWrapperTest.scala
index 7d25cb1..9a63abd 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/context/MetricWrapperTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/context/MetricWrapperTest.scala
@@ -17,9 +17,10 @@
package org.apache.griffin.measure.context
-import org.scalatest._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
-class MetricWrapperTest extends FlatSpec with Matchers {
+class MetricWrapperTest extends AnyFlatSpec with Matchers {
"metric wrapper" should "flush empty if no metric inserted" in {
val metricWrapper = MetricWrapper("name", "appId")
diff --git a/measure/src/test/scala/org/apache/griffin/measure/context/TimeRangeTest.scala b/measure/src/test/scala/org/apache/griffin/measure/context/TimeRangeTest.scala
index 28c195b..9648d96 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/context/TimeRangeTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/context/TimeRangeTest.scala
@@ -17,9 +17,10 @@
package org.apache.griffin.measure.context
-import org.scalatest._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
-class TimeRangeTest extends FlatSpec with Matchers {
+class TimeRangeTest extends AnyFlatSpec with Matchers {
"time range" should "be able to merge another time range" in {
val tr1 = TimeRange(1, 10, Set(2, 5, 8))
diff --git a/measure/src/test/scala/org/apache/griffin/measure/datasource/TimestampStorageTest.scala b/measure/src/test/scala/org/apache/griffin/measure/datasource/TimestampStorageTest.scala
index a90768e..c60b439 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/datasource/TimestampStorageTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/datasource/TimestampStorageTest.scala
@@ -17,9 +17,10 @@
package org.apache.griffin.measure.datasource
-import org.scalatest._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
-class TimestampStorageTest extends FlatSpec with Matchers {
+class TimestampStorageTest extends AnyFlatSpec with Matchers {
"timestamp storage" should "be able to insert a timestamp" in {
val timestampStorage = TimestampStorage()
diff --git a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactorySpec.scala b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactorySpec.scala
index 466196b..fd4e6a5 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactorySpec.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/DataConnectorFactorySpec.scala
@@ -23,7 +23,7 @@
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.dstream.InputDStream
-import org.scalatest.FlatSpec
+import org.scalatest.flatspec.AnyFlatSpec
import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.context.TimeRange
@@ -75,7 +75,7 @@
override def data(ms: Long): (Option[DataFrame], TimeRange) = null
}
-class DataConnectorFactorySpec extends FlatSpec {
+class DataConnectorFactorySpec extends AnyFlatSpec {
"DataConnectorFactory" should "be able to create custom batch connector" in {
val param = DataConnectorParam(
diff --git a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchDataConnectorTest.scala b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchDataConnectorTest.scala
index 60c50ca..37c63fc 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchDataConnectorTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/ElasticSearchDataConnectorTest.scala
@@ -18,13 +18,15 @@
package org.apache.griffin.measure.datasource.connector.batch
import org.apache.spark.sql.types.StructType
-import org.scalatest.Matchers
+import org.scalatest.matchers.should._
+import org.scalatest.Ignore
import org.testcontainers.elasticsearch.ElasticsearchContainer
import org.apache.griffin.measure.SparkSuiteBase
import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
import org.apache.griffin.measure.datasource.TimestampStorage
+@Ignore
class ElasticSearchDataConnectorTest extends SparkSuiteBase with Matchers {
// ignorance flag that could skip cases
diff --git a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnectorTest.scala b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnectorTest.scala
index 2493df3..2ae65be 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnectorTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/FileBasedDataConnectorTest.scala
@@ -18,7 +18,7 @@
package org.apache.griffin.measure.datasource.connector.batch
import org.apache.spark.sql.types.{IntegerType, LongType, StringType, StructType}
-import org.scalatest._
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.SparkSuiteBase
import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
@@ -65,13 +65,12 @@
// valid schema
val result1 = FileBasedDataConnector(
spark,
- dcParam.copy(
- config = configs + (
- (
- "schema",
- Seq(
- Map("name" -> "name", "type" -> "string"),
- Map("name" -> "age", "type" -> "int", "nullable" -> "true"))))),
+ dcParam.copy(config = configs + (
+ (
+ "schema",
+ Seq(
+ Map("name" -> "name", "type" -> "string"),
+ Map("name" -> "age", "type" -> "int", "nullable" -> "true"))))),
timestampStorage)
.data(1L)
diff --git a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnectorTest.scala b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnectorTest.scala
index 77040e3..8a755b6 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnectorTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/datasource/connector/batch/JDBCBasedDataConnectorTest.scala
@@ -19,7 +19,7 @@
import java.sql.DriverManager
import java.util.Properties
-import org.scalatest.Matchers
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.SparkSuiteBase
import org.apache.griffin.measure.configuration.dqdefinition.DataConnectorParam
diff --git a/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala b/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala
index a557dda..9fc9883 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/job/DQAppTest.scala
@@ -19,7 +19,7 @@
import scala.util.{Failure, Success}
-import org.scalatest.{BeforeAndAfterAll, FlatSpec, Matchers}
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.{Loggable, SparkSuiteBase}
import org.apache.griffin.measure.Application._
@@ -30,12 +30,7 @@
import org.apache.griffin.measure.launch.batch.BatchDQApp
import org.apache.griffin.measure.launch.streaming.StreamingDQApp
-class DQAppTest
- extends FlatSpec
- with SparkSuiteBase
- with BeforeAndAfterAll
- with Matchers
- with Loggable {
+class DQAppTest extends SparkSuiteBase with Matchers with Loggable {
var envParam: EnvConfig = _
var sparkParam: SparkParam = _
diff --git a/measure/src/test/scala/org/apache/griffin/measure/sink/CustomSinkTest.scala b/measure/src/test/scala/org/apache/griffin/measure/sink/CustomSinkTest.scala
index e4754e0..8bf81b1 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/sink/CustomSinkTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/sink/CustomSinkTest.scala
@@ -57,7 +57,7 @@
}
sinks.headOption match {
case Some(sink: CustomSink) => sink.allMetrics
- case _ => mutable.ListBuffer[String]()
+ case _ => Map.empty
}
})
diff --git a/measure/src/test/scala/org/apache/griffin/measure/sink/SinkTestBase.scala b/measure/src/test/scala/org/apache/griffin/measure/sink/SinkTestBase.scala
index 919183b..b13858f 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/sink/SinkTestBase.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/sink/SinkTestBase.scala
@@ -19,14 +19,15 @@
import org.apache.spark.sql.{DataFrame, Row}
import org.apache.spark.sql.types._
-import org.scalatest.{FlatSpec, Matchers}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.{Loggable, SparkSuiteBase}
import org.apache.griffin.measure.configuration.dqdefinition.SinkParam
import org.apache.griffin.measure.configuration.enums.ProcessType.BatchProcessType
import org.apache.griffin.measure.context.{ContextId, DQContext}
-trait SinkTestBase extends FlatSpec with Matchers with SparkSuiteBase with Loggable {
+trait SinkTestBase extends AnyFlatSpec with Matchers with SparkSuiteBase with Loggable {
var sinkParams: Seq[SinkParam]
diff --git a/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala b/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala
index 834d8e0..5892c11 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/step/TransformStepTest.scala
@@ -17,15 +17,17 @@
package org.apache.griffin.measure.step
-import org.scalatest._
import scala.util.Try
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
+
import org.apache.griffin.measure.{Loggable, SparkSuiteBase}
import org.apache.griffin.measure.configuration.enums.ProcessType.BatchProcessType
import org.apache.griffin.measure.context.{ContextId, DQContext}
import org.apache.griffin.measure.step.transform.TransformStep
-class TransformStepTest extends FlatSpec with Matchers with SparkSuiteBase with Loggable {
+class TransformStepTest extends AnyFlatSpec with Matchers with SparkSuiteBase with Loggable {
case class DualTransformStep(
name: String,
diff --git a/measure/src/test/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQStepsTest.scala b/measure/src/test/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQStepsTest.scala
index 67e5236..f80fb13 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQStepsTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/step/builder/dsl/transform/CompletenessExpr2DQStepsTest.scala
@@ -17,15 +17,15 @@
package org.apache.griffin.measure.step.builder.dsl.transform
-import org.scalatest._
-import org.scalatest.mockito.MockitoSugar
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
+import org.scalatestplus.mockito.MockitoSugar
-import org.apache.griffin.measure.configuration.dqdefinition.RuleErrorConfParam
-import org.apache.griffin.measure.configuration.dqdefinition.RuleParam
+import org.apache.griffin.measure.configuration.dqdefinition.{RuleErrorConfParam, RuleParam}
import org.apache.griffin.measure.context.DQContext
import org.apache.griffin.measure.step.builder.dsl.expr.Expr
-class CompletenessExpr2DQStepsTest extends FlatSpec with Matchers with MockitoSugar {
+class CompletenessExpr2DQStepsTest extends AnyFlatSpec with Matchers with MockitoSugar {
"CompletenessExpr2DQSteps" should "get correct where clause" in {
val completeness = CompletenessExpr2DQSteps(mock[DQContext], mock[Expr], mock[RuleParam])
diff --git a/measure/src/test/scala/org/apache/griffin/measure/transformations/AccuracyTransformationsIntegrationTest.scala b/measure/src/test/scala/org/apache/griffin/measure/transformations/AccuracyTransformationsIntegrationTest.scala
index 2f06ad3..5b019a5 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/transformations/AccuracyTransformationsIntegrationTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/transformations/AccuracyTransformationsIntegrationTest.scala
@@ -18,7 +18,8 @@
package org.apache.griffin.measure.transformations
import org.apache.spark.sql.DataFrame
-import org.scalatest._
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.SparkSuiteBase
import org.apache.griffin.measure.configuration.dqdefinition._
@@ -29,7 +30,10 @@
case class AccuracyResult(total: Long, miss: Long, matched: Long, matchedFraction: Double)
-class AccuracyTransformationsIntegrationTest extends FlatSpec with Matchers with SparkSuiteBase {
+class AccuracyTransformationsIntegrationTest
+ extends AnyFlatSpec
+ with Matchers
+ with SparkSuiteBase {
private val EMPTY_PERSON_TABLE = "empty_person"
private val PERSON_TABLE = "person"
@@ -80,10 +84,9 @@
sourceName: String,
targetName: String,
expectedResult: AccuracyResult) = {
- val dqContext: DQContext = getDqContext(
- dataSourcesParam = List(
- DataSourceParam(name = "source", connector = dataConnectorParam(tableName = sourceName)),
- DataSourceParam(name = "target", connector = dataConnectorParam(tableName = targetName))))
+ val dqContext: DQContext = getDqContext(dataSourcesParam = List(
+ DataSourceParam(name = "source", connector = dataConnectorParam(tableName = sourceName)),
+ DataSourceParam(name = "target", connector = dataConnectorParam(tableName = targetName))))
val accuracyRule = RuleParam(
dslType = "griffin-dsl",
diff --git a/measure/src/test/scala/org/apache/griffin/measure/utils/ParamUtilTest.scala b/measure/src/test/scala/org/apache/griffin/measure/utils/ParamUtilTest.scala
index 720f9b2..9f204a8 100644
--- a/measure/src/test/scala/org/apache/griffin/measure/utils/ParamUtilTest.scala
+++ b/measure/src/test/scala/org/apache/griffin/measure/utils/ParamUtilTest.scala
@@ -17,11 +17,13 @@
package org.apache.griffin.measure.utils
-import org.scalatest._
+import org.scalatest.BeforeAndAfter
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should._
import org.apache.griffin.measure.utils.ParamUtil._
-class ParamUtilTest extends FlatSpec with Matchers with BeforeAndAfter {
+class ParamUtilTest extends AnyFlatSpec with Matchers with BeforeAndAfter {
val fruits: Map[String, Any] =
Map[String, Any]("A" -> "apple", "B" -> "banana", "O" -> "orange")
diff --git a/pom.xml b/pom.xml
index 7a68625..e51c94e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -30,7 +30,7 @@
<groupId>org.apache.griffin</groupId>
<artifactId>griffin</artifactId>
- <version>0.6.0-SNAPSHOT</version>
+ <version>0.7.0-SNAPSHOT</version>
<packaging>pom</packaging>
<name>Apache Griffin ${project.version}</name>
<url>http://griffin.apache.org</url>
@@ -40,13 +40,23 @@
</prerequisites>
<properties>
+ <encoding>UTF-8</encoding>
+ <project.build.sourceEncoding>${encoding}</project.build.sourceEncoding>
+ <project.reporting.outputEncoding>${encoding}</project.reporting.outputEncoding>
+
<java.version>1.8</java.version>
+ <scala.binary.version>2.11</scala.binary.version>
+ <scala211.binary.version>2.11</scala211.binary.version>
+ <scala.version>${scala.binary.version}.0</scala.version>
+
+ <maven.compiler.source>${java.version}</maven.compiler.source>
+ <maven.compiler.target>${java.version}</maven.compiler.target>
+
<maven-apache-rat.version>0.11</maven-apache-rat.version>
<exec-maven-plugin.version>1.6.0</exec-maven-plugin.version>
- <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
- <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
- <maven.compiler.source>1.8</maven.compiler.source>
- <maven.compiler.target>1.8</maven.compiler.target>
+
+ <compile.scope>compile</compile.scope>
+ <provided.scope>provided</provided.scope>
</properties>
<licenses>
@@ -109,7 +119,7 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
- <version>3.6.1</version>
+ <version>3.8.1</version>
<configuration>
<source>${maven.compiler.source}</source>
<target>${maven.compiler.target}</target>
@@ -119,10 +129,16 @@
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-surefire-plugin</artifactId>
- <version>2.22.0</version>
+ <version>3.0.0-M3</version>
<configuration>
+ <testFailureIgnore>false</testFailureIgnore>
+ <useSystemClassLoader>true</useSystemClassLoader>
+ <forkMode>once</forkMode>
+ <failIfNoTests>false</failIfNoTests>
<reportsDirectory>${project.build.directory}/surefire-reports</reportsDirectory>
- <argLine>-ea -Xmx1g -Xss4m -XX:ReservedCodeCacheSize=128m</argLine>
+ <systemProperties>
+ <java.io.tmpdir>${project.build.directory}/tmp</java.io.tmpdir>
+ </systemProperties>
</configuration>
<executions>
<execution>
@@ -133,24 +149,7 @@
</execution>
</executions>
</plugin>
- <!-- Scalatest runs all Scala tests -->
- <!-- enable scalatest -->
- <plugin>
- <groupId>org.scalatest</groupId>
- <artifactId>scalatest-maven-plugin</artifactId>
- <version>1.0</version>
- <configuration>
- <argLine>-ea -Xmx1g -Xss4m -XX:ReservedCodeCacheSize=128m</argLine>
- </configuration>
- <executions>
- <execution>
- <id>test</id>
- <goals>
- <goal>test</goal>
- </goals>
- </execution>
- </executions>
- </plugin>
+
<plugin>
<groupId>org.apache.rat</groupId>
<artifactId>apache-rat-plugin</artifactId>
@@ -196,6 +195,7 @@
<exclude>**/src/main/resources/public/**</exclude>
<exclude>**/pom.xml.releaseBackup</exclude>
<exclude>**/pom.xml.tag</exclude>
+ <exclude>**/_SUCCESS</exclude>
</excludes>
</configuration>
<executions>
diff --git a/service/pom.xml b/service/pom.xml
index dd2466f..8121998 100644
--- a/service/pom.xml
+++ b/service/pom.xml
@@ -23,7 +23,7 @@
<parent>
<groupId>org.apache.griffin</groupId>
<artifactId>griffin</artifactId>
- <version>0.6.0-SNAPSHOT</version>
+ <version>0.7.0-SNAPSHOT</version>
</parent>
<artifactId>service</artifactId>
@@ -34,7 +34,6 @@
<properties>
<hadoop.version>2.7.1</hadoop.version>
<hive.version>2.2.0</hive.version>
- <scala.version>2.10</scala.version>
<spring.boot.version>2.1.7.RELEASE</spring.boot.version>
<spring.security.kerberos.version>1.0.1.RELEASE</spring.security.kerberos.version>
<confluent.version>3.2.0</confluent.version>
@@ -258,7 +257,7 @@
<!--livy-core-->
<dependency>
<groupId>com.cloudera.livy</groupId>
- <artifactId>livy-core_${scala.version}</artifactId>
+ <artifactId>livy-core_2.10</artifactId>
<version>${livy.core.version}</version>
</dependency>
diff --git a/service/src/main/java/org/apache/griffin/core/job/BatchJobOperatorImpl.java b/service/src/main/java/org/apache/griffin/core/job/BatchJobOperatorImpl.java
index b46ea8b..9eb60ad 100644
--- a/service/src/main/java/org/apache/griffin/core/job/BatchJobOperatorImpl.java
+++ b/service/src/main/java/org/apache/griffin/core/job/BatchJobOperatorImpl.java
@@ -403,7 +403,7 @@
Set<String> sets = new HashSet<>();
List<DataSource> sources = measure.getDataSources();
for (DataSource source : sources) {
- source.getConnectors().forEach(dc -> sets.add(dc.getName()));
+ sets.add(source.getConnector().getName());
}
if (sets.size() < sources.size()) {
LOGGER.warn("Connector names cannot be repeated.");
diff --git a/service/src/main/java/org/apache/griffin/core/job/JobInstance.java b/service/src/main/java/org/apache/griffin/core/job/JobInstance.java
index 3899147..344cf75 100644
--- a/service/src/main/java/org/apache/griffin/core/job/JobInstance.java
+++ b/service/src/main/java/org/apache/griffin/core/job/JobInstance.java
@@ -146,8 +146,7 @@
jobStartTime = triggerTime.getTime();
}
- private void setSourcesPartitionsAndPredicates(List<DataSource> sources)
- throws Exception {
+ private void setSourcesPartitionsAndPredicates(List<DataSource> sources) {
boolean isFirstBaseline = true;
for (JobDataSegment jds : job.getSegments()) {
if (jds.isAsTsBaseline() && isFirstBaseline) {
@@ -157,22 +156,14 @@
isFirstBaseline = false;
}
for (DataSource ds : sources) {
- setDataSourcePartitions(jds, ds);
+ setDataConnectorPartitions(jds, ds.getConnector());
}
}
}
- private void setDataSourcePartitions(JobDataSegment jds, DataSource ds)
- throws Exception {
- List<DataConnector> connectors = ds.getConnectors();
- for (DataConnector dc : connectors) {
- setDataConnectorPartitions(jds, dc);
- }
- }
-
private void setDataConnectorPartitions(
JobDataSegment jds,
- DataConnector dc) throws Exception {
+ DataConnector dc) {
String dcName = jds.getDataConnectorName();
if (dcName.equals(dc.getName())) {
Long[] sampleTs = genSampleTs(jds.getSegmentRange(), dc);
diff --git a/service/src/main/java/org/apache/griffin/core/measure/entity/DataSource.java b/service/src/main/java/org/apache/griffin/core/measure/entity/DataSource.java
index 970977b..378b397 100644
--- a/service/src/main/java/org/apache/griffin/core/measure/entity/DataSource.java
+++ b/service/src/main/java/org/apache/griffin/core/measure/entity/DataSource.java
@@ -28,21 +28,10 @@
import java.io.IOException;
import java.util.ArrayList;
-import java.util.List;
import java.util.Map;
-import javax.persistence.CascadeType;
-import javax.persistence.Column;
-import javax.persistence.Entity;
-import javax.persistence.FetchType;
-import javax.persistence.JoinColumn;
-import javax.persistence.OneToMany;
-import javax.persistence.PostLoad;
-import javax.persistence.PrePersist;
-import javax.persistence.PreUpdate;
-import javax.persistence.Transient;
+import javax.persistence.*;
import org.apache.griffin.core.util.JsonUtil;
-import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;
@Entity
@@ -51,10 +40,10 @@
private String name;
- @OneToMany(fetch = FetchType.EAGER, cascade = {CascadeType.PERSIST,
+ @OneToOne(fetch = FetchType.EAGER, cascade = {CascadeType.PERSIST,
CascadeType.REMOVE, CascadeType.MERGE})
@JoinColumn(name = "data_source_id")
- private List<DataConnector> connectors = new ArrayList<>();
+ private DataConnector connector = new DataConnector();
private boolean baseline = false;
@@ -75,15 +64,12 @@
this.name = name;
}
- public List<DataConnector> getConnectors() {
- return connectors;
+ public DataConnector getConnector() {
+ return connector;
}
- public void setConnectors(List<DataConnector> connectors) {
- if (CollectionUtils.isEmpty(connectors)) {
- throw new NullPointerException("Data connector can not be empty.");
- }
- this.connectors = connectors;
+ public void setConnector(DataConnector connector) {
+ this.connector = connector;
}
public boolean isBaseline() {
@@ -132,18 +118,18 @@
public DataSource() {
}
- public DataSource(String name, List<DataConnector> connectors) {
+ public DataSource(String name, DataConnector connector) {
this.name = name;
- this.connectors = connectors;
+ this.connector = connector;
}
public DataSource(String name, boolean baseline,
Map<String, Object> checkpointMap,
- List<DataConnector> connectors) {
+ DataConnector connector) {
this.name = name;
this.baseline = baseline;
this.checkpointMap = checkpointMap;
- this.connectors = connectors;
+ this.connector = connector;
}
}
diff --git a/service/src/main/java/org/apache/griffin/core/metastore/hive/HiveMetaStoreServiceImpl.java b/service/src/main/java/org/apache/griffin/core/metastore/hive/HiveMetaStoreServiceImpl.java
index d855183..4b83907 100644
--- a/service/src/main/java/org/apache/griffin/core/metastore/hive/HiveMetaStoreServiceImpl.java
+++ b/service/src/main/java/org/apache/griffin/core/metastore/hive/HiveMetaStoreServiceImpl.java
@@ -49,7 +49,7 @@
private static final Logger LOGGER = LoggerFactory
.getLogger(HiveMetaStoreService.class);
- @Autowired
+ @Autowired(required = false)
private IMetaStoreClient client = null;
@Value("${hive.metastore.dbname}")
diff --git a/service/src/main/java/org/apache/griffin/core/util/MeasureUtil.java b/service/src/main/java/org/apache/griffin/core/util/MeasureUtil.java
index 7e57d87..59526bd 100644
--- a/service/src/main/java/org/apache/griffin/core/util/MeasureUtil.java
+++ b/service/src/main/java/org/apache/griffin/core/util/MeasureUtil.java
@@ -42,7 +42,7 @@
public class MeasureUtil {
private static final Logger LOGGER = LoggerFactory
- .getLogger(MeasureUtil.class);
+ .getLogger(MeasureUtil.class);
public static void validateMeasure(Measure measure) {
if (measure instanceof GriffinMeasure) {
@@ -56,7 +56,7 @@
private static void validateGriffinMeasure(GriffinMeasure measure) {
if (getConnectorNamesIfValid(measure) == null) {
throw new GriffinException.BadRequestException
- (INVALID_CONNECTOR_NAME);
+ (INVALID_CONNECTOR_NAME);
}
if (!validatePredicates(measure)) {
throw new GriffinException.BadRequestException(INVALID_MEASURE_PREDICATE);
@@ -65,13 +65,11 @@
private static boolean validatePredicates(GriffinMeasure measure) {
for (DataSource dataSource : measure.getDataSources()) {
- for (DataConnector dataConnector : dataSource.getConnectors()) {
- for (SegmentPredicate segmentPredicate : dataConnector.getPredicates()) {
- try {
- PredicatorFactory.newPredicateInstance(segmentPredicate);
- } catch (Exception e) {
- return false;
- }
+ for (SegmentPredicate segmentPredicate : dataSource.getConnector().getPredicates()) {
+ try {
+ PredicatorFactory.newPredicateInstance(segmentPredicate);
+ } catch (Exception e) {
+ return false;
}
}
}
@@ -81,7 +79,7 @@
private static void validateExternalMeasure(ExternalMeasure measure) {
if (StringUtils.isBlank(measure.getMetricName())) {
LOGGER.warn("Failed to create external measure {}. " +
- "Its metric name is blank.", measure.getName());
+ "Its metric name is blank.", measure.getName());
throw new GriffinException.BadRequestException(MISSING_METRIC_NAME);
}
}
@@ -90,8 +88,9 @@
Set<String> sets = new HashSet<>();
List<DataSource> sources = measure.getDataSources();
for (DataSource source : sources) {
- source.getConnectors().stream().filter(dc -> dc.getName() != null)
- .forEach(dc -> sets.add(dc.getName()));
+ if(source.getConnector() != null && source.getConnector().getName() != null){
+ sets.add(source.getConnector().getName());
+ }
}
if (sets.size() == 0 || sets.size() < sources.size()) {
LOGGER.warn("Connector names cannot be repeated or empty.");
diff --git a/service/src/main/resources/env/env_batch.json b/service/src/main/resources/env/env_batch.json
index 72a3839..9ed9ef7 100644
--- a/service/src/main/resources/env/env_batch.json
+++ b/service/src/main/resources/env/env_batch.json
@@ -4,20 +4,23 @@
},
"sinks": [
{
+ "name": "console",
"type": "CONSOLE",
"config": {
"max.log.lines": 10
}
},
{
+ "name": "hdfs",
"type": "HDFS",
"config": {
- "path": "hdfs:///griffin/persist",
+ "path": "hdfs://localhost/griffin/persist",
"max.persist.lines": 10000,
"max.lines.per.file": 10000
}
},
{
+ "name": "elasticsearch",
"type": "ELASTICSEARCH",
"config": {
"method": "post",
diff --git a/service/src/main/resources/log4j2-spring.xml b/service/src/main/resources/log4j2-spring.xml
index c021b4a..9cee7a5 100644
--- a/service/src/main/resources/log4j2-spring.xml
+++ b/service/src/main/resources/log4j2-spring.xml
@@ -1,4 +1,20 @@
<?xml version="1.0" encoding="UTF-8"?>
+<?xml-stylesheet type="text/xsl" href="configuration.xsl"?><!--
+ 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.
+-->
<Configuration status="WARN">
<Properties>
<Property name="PID">????</Property>
diff --git a/service/src/test/java/org/apache/griffin/core/measure/repo/MeasureRepoTest.java b/service/src/test/java/org/apache/griffin/core/measure/repo/MeasureRepoTest.java
index 00ce1fa..534527f 100644
--- a/service/src/test/java/org/apache/griffin/core/measure/repo/MeasureRepoTest.java
+++ b/service/src/test/java/org/apache/griffin/core/measure/repo/MeasureRepoTest.java
@@ -66,7 +66,7 @@
GriffinMeasure m = (GriffinMeasure) measures.get(0);
List<DataSource> sources = m.getDataSources();
- DataConnector connector = sources.get(0).getConnectors().get(0);
+ DataConnector connector = sources.get(0).getConnector();
Rule rule = m.getEvaluateRule().getRules().get(0);
assertEquals(m.getSinksList().size(), 2);
assertEquals(sources.get(0).isBaseline(), true);
diff --git a/service/src/test/java/org/apache/griffin/core/util/EntityMocksHelper.java b/service/src/test/java/org/apache/griffin/core/util/EntityMocksHelper.java
index 563210d..6d9f053 100644
--- a/service/src/test/java/org/apache/griffin/core/util/EntityMocksHelper.java
+++ b/service/src/test/java/org/apache/griffin/core/util/EntityMocksHelper.java
@@ -82,9 +82,9 @@
DataConnector dcTarget)
throws Exception {
DataSource dataSource = new DataSource(
- "source", true, createCheckpointMap(), Arrays.asList(dcSource));
+ "source", true, createCheckpointMap(), dcSource);
DataSource targetSource = new DataSource(
- "target", false, createCheckpointMap(), Arrays.asList(dcTarget));
+ "target", false, createCheckpointMap(), dcTarget);
List<DataSource> dataSources = new ArrayList<>();
dataSources.add(dataSource);
dataSources.add(targetSource);
diff --git a/ui/angular/package.json b/ui/angular/package.json
index d4fae3d..39c6327 100644
--- a/ui/angular/package.json
+++ b/ui/angular/package.json
@@ -39,7 +39,7 @@
"zone.js": "^0.8.14"
},
"devDependencies": {
- "@angular/cli": "1.3.0",
+ "@angular/cli": "1.7.4",
"@angular/compiler-cli": "4.4.4",
"@angular/language-service": "4.4.4",
"@types/jasmine": "~2.5.53",
diff --git a/ui/angular/src/app/job/create-job/batch/batch.component.ts b/ui/angular/src/app/job/create-job/batch/batch.component.ts
index 150f307..9049761 100644
--- a/ui/angular/src/app/job/create-job/batch/batch.component.ts
+++ b/ui/angular/src/app/job/create-job/batch/batch.component.ts
@@ -268,23 +268,21 @@
if (measure == map.name) {
var source = map["data.sources"];
for (let i = 0; i < source.length; i++) {
- var details = source[i].connectors;
- for (let j = 0; j < details.length; j++) {
- if (details[j]["data.unit"] != undefined) {
- var table =
- details[j].config.database +
- "." +
- details[j].config["table.name"];
- var size = details[j]["data.unit"];
- var connectorname = details[j]["name"];
- var detail = {
- id: i + 1,
- name: table,
- size: size,
- connectorname: connectorname
- };
- this.dropdownList.push(detail);
- }
+ var connector = source[i].connector;
+ if (connector["data.unit"] != undefined) {
+ var table =
+ connector.config.database +
+ "." +
+ connector.config["table.name"];
+ var size = connector["data.unit"];
+ var connectorname = connector["name"];
+ var detail = {
+ id: i + 1,
+ name: table,
+ size: size,
+ connectorname: connectorname
+ };
+ this.dropdownList.push(detail);
}
}
}
diff --git a/ui/angular/src/app/job/create-job/streaming/streaming.component.ts b/ui/angular/src/app/job/create-job/streaming/streaming.component.ts
index 200d788..808631b 100644
--- a/ui/angular/src/app/job/create-job/streaming/streaming.component.ts
+++ b/ui/angular/src/app/job/create-job/streaming/streaming.component.ts
@@ -234,23 +234,21 @@
if (measure == map.name) {
var source = map["data.sources"];
for (let i = 0; i < source.length; i++) {
- var details = source[i].connectors;
- for (let j = 0; j < details.length; j++) {
- if (details[j]["data.unit"] != undefined) {
- var table =
- details[j].config.database +
- "." +
- details[j].config["table.name"];
- var size = details[j]["data.unit"];
- var connectorname = details[j]["name"];
- var detail = {
- id: i + 1,
- name: table,
- size: size,
- connectorname: connectorname
- };
- this.dropdownList.push(detail);
- }
+ var connector = source[i].connector;
+ if (connector["data.unit"] != undefined) {
+ var table =
+ connector.config.database +
+ "." +
+ connector.config["table.name"];
+ var size = connector["data.unit"];
+ var connectorname = connector["name"];
+ var detail = {
+ id: i + 1,
+ name: table,
+ size: size,
+ connectorname: connectorname
+ };
+ this.dropdownList.push(detail);
}
}
}
diff --git a/ui/angular/src/app/job/job-detail/job-detail.component.ts b/ui/angular/src/app/job/job-detail/job-detail.component.ts
index 7c4a10b..d857b26 100644
--- a/ui/angular/src/app/job/job-detail/job-detail.component.ts
+++ b/ui/angular/src/app/job/job-detail/job-detail.component.ts
@@ -57,7 +57,7 @@
this.measureType = this.measureData["dq.type"].toLowerCase();
this.processType = this.measureData["process.type"].toLowerCase();
for (let item of this.measureData["data.sources"]) {
- let config = item.connectors[0].config;
+ let config = item.connector.config;
let tableName = config.database + "." + config["table.name"];
this.tableInfo.push(tableName);
}
diff --git a/ui/angular/src/app/measure/create-measure/ac/ac.component.ts b/ui/angular/src/app/measure/create-measure/ac/ac.component.ts
index 72d0138..8b7bbec 100644
--- a/ui/angular/src/app/measure/create-measure/ac/ac.component.ts
+++ b/ui/angular/src/app/measure/create-measure/ac/ac.component.ts
@@ -148,7 +148,7 @@
"data.sources": [
{
name: "source",
- connectors: [
+ connector:
{
name: "",
type: "HIVE",
@@ -170,11 +170,10 @@
}
]
}
- ]
},
{
name: "target",
- connectors: [
+ connector:
{
name: "",
type: "HIVE",
@@ -196,7 +195,6 @@
}
]
}
- ]
}
],
@@ -393,7 +391,7 @@
"data.sources": [
{
name: "source",
- connectors: [
+ connector:
{
name: this.src_name,
type: "HIVE",
@@ -415,11 +413,10 @@
}
]
}
- ]
},
{
name: "target",
- connectors: [
+ connector:
{
name: this.tgt_name,
type: "HIVE",
@@ -441,7 +438,6 @@
}
]
}
- ]
}
],
"evaluate.rule": {
@@ -499,11 +495,11 @@
}
deleteUnit(index) {
- delete this.newMeasure["data.sources"][index]["connectors"][0]["data.unit"];
+ delete this.newMeasure["data.sources"][index]["connector"]["data.unit"];
}
deletePredicates(index) {
- delete this.newMeasure["data.sources"][index]["connectors"][0]["predicates"];
+ delete this.newMeasure["data.sources"][index]["connector"]["predicates"];
}
save() {
diff --git a/ui/angular/src/app/measure/create-measure/pr/pr.component.ts b/ui/angular/src/app/measure/create-measure/pr/pr.component.ts
index a1dae5e..57be58f 100644
--- a/ui/angular/src/app/measure/create-measure/pr/pr.component.ts
+++ b/ui/angular/src/app/measure/create-measure/pr/pr.component.ts
@@ -360,7 +360,7 @@
"data.sources": [
{
name: "source",
- connectors: [
+ connector:
{
name: this.step1.srcname,
type: "HIVE",
@@ -382,7 +382,6 @@
}
]
}
- ]
}
],
"evaluate.rule": {
@@ -393,10 +392,10 @@
this.getGrouprule();
if (this.step3.size.indexOf("0") == 0) {
- delete this.newMeasure["data.sources"][0]["connectors"][0]["data.unit"];
+ delete this.newMeasure["data.sources"][0]["connector"]["data.unit"];
}
if (!this.step3.needpath || this.step3.path == "") {
- delete this.newMeasure["data.sources"][0]["connectors"][0]["predicates"];
+ delete this.newMeasure["data.sources"][0]["connector"]["predicates"];
}
this.visible = true;
setTimeout(() => (this.visibleAnimate = true), 100);
diff --git a/ui/angular/src/app/measure/measure-detail/measure-detail.component.spec.ts b/ui/angular/src/app/measure/measure-detail/measure-detail.component.spec.ts
index 6fd2f25..08d9dc4 100644
--- a/ui/angular/src/app/measure/measure-detail/measure-detail.component.spec.ts
+++ b/ui/angular/src/app/measure/measure-detail/measure-detail.component.spec.ts
@@ -43,7 +43,7 @@
});
it(
- 'should be created',
+ 'should be created',
inject(
[HttpTestingController, ServiceService],
(httpMock: HttpTestingController, serviceService: ServiceService) => {
@@ -54,13 +54,13 @@
"dq.type": "",
"evaluate.rule": "",
"data.sources": [{
- "connectors": [{
+ "connector": {
"data.unit": "",
config: {
where: ""
},
predicates: []
- }]
+ }
}]
});
diff --git a/ui/angular/src/app/measure/measure-detail/measure-detail.component.ts b/ui/angular/src/app/measure/measure-detail/measure-detail.component.ts
index 5ad5fae..7413af1 100644
--- a/ui/angular/src/app/measure/measure-detail/measure-detail.component.ts
+++ b/ui/angular/src/app/measure/measure-detail/measure-detail.component.ts
@@ -64,7 +64,7 @@
currentrule: string;
fetchData(value, index) {
- var data = this.ruleData["data.sources"][index].connectors[0];
+ var data = this.ruleData["data.sources"][index].connector;
var size = value + "size";
var zone = value + "zone";
var where = value + "where";
diff --git a/ui/angular/src/app/measure/measure.component.ts b/ui/angular/src/app/measure/measure.component.ts
index 23f7a15..d94dc3c 100644
--- a/ui/angular/src/app/measure/measure.component.ts
+++ b/ui/angular/src/app/measure/measure.component.ts
@@ -72,11 +72,11 @@
this.deletedRow = row;
$("#save").removeAttr("disabled");
if (this.deletedRow["measure.type"] !== "external") {
- var sourcedata = this.deletedRow["data.sources"][0].connectors[0].config;
+ var sourcedata = this.deletedRow["data.sources"][0].connector.config;
this.sourceTable = sourcedata["table.name"];
}
if (this.deletedRow["dq.type"] === "accuracy") {
- var targetdata = this.deletedRow["data.sources"][1].connectors[0].config;
+ var targetdata = this.deletedRow["data.sources"][1].connector.config;
this.targetTable = targetdata["table.name"];
} else {
this.targetTable = "";
diff --git a/ui/pom.xml b/ui/pom.xml
index d42cb6a..465da85 100644
--- a/ui/pom.xml
+++ b/ui/pom.xml
@@ -24,7 +24,7 @@
<parent>
<groupId>org.apache.griffin</groupId>
<artifactId>griffin</artifactId>
- <version>0.6.0-SNAPSHOT</version>
+ <version>0.7.0-SNAPSHOT</version>
</parent>
<artifactId>ui</artifactId>
<packaging>pom</packaging>
