You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@predictionio.apache.org by do...@apache.org on 2017/03/16 14:52:14 UTC
[1/7] incubator-predictionio git commit: Unit test script should fail
immediately
Repository: incubator-predictionio
Updated Branches:
refs/heads/feature/xbuild b6c90a0ce -> d051216a7
Unit test script should fail immediately
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/606be414
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/606be414
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/606be414
Branch: refs/heads/feature/xbuild
Commit: 606be4148fe7d4253193a4ef3c4e996d27c0ad51
Parents: ae51040
Author: Donald Szeto <do...@apache.org>
Authored: Wed Mar 15 09:40:26 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Wed Mar 15 09:40:26 2017 -0700
----------------------------------------------------------------------
tests/unit.sh | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/606be414/tests/unit.sh
----------------------------------------------------------------------
diff --git a/tests/unit.sh b/tests/unit.sh
index 324b87f..6382a70 100755
--- a/tests/unit.sh
+++ b/tests/unit.sh
@@ -1,4 +1,4 @@
-#!/bin/bash
+#!/bin/bash -ex
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
[5/7] incubator-predictionio git commit: Merge branch 'develop' into
feature/xbuild
Posted by do...@apache.org.
Merge branch 'develop' into feature/xbuild
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/ab119806
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/ab119806
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/ab119806
Branch: refs/heads/feature/xbuild
Commit: ab119806ee7532b9baa09070813d23c4597d5c10
Parents: be8397f bb00324
Author: Donald Szeto <do...@apache.org>
Authored: Thu Mar 16 00:13:30 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Thu Mar 16 00:13:30 2017 -0700
----------------------------------------------------------------------
.../predictionio/data/storage/LEventsSpec.scala | 248 -----------------
.../predictionio/data/storage/PEventsSpec.scala | 213 ---------------
.../data/storage/StorageTestUtils.scala | 45 ----
docs/manual/source/templates/index.html.md | 25 +-
.../data/storage/hbase/LEventsSpec.scala | 239 +++++++++++++++++
.../data/storage/hbase/PEventsSpec.scala | 192 +++++++++++++
.../data/storage/hbase/StorageTestUtils.scala | 40 +++
.../data/storage/hbase/TestEvents.scala | 266 +++++++++++++++++++
.../data/storage/jdbc/LEventsSpec.scala | 236 ++++++++++++++++
.../data/storage/jdbc/PEventsSpec.scala | 193 ++++++++++++++
.../data/storage/jdbc/StorageTestUtils.scala | 29 ++
.../data/storage/jdbc/TestEvents.scala | 266 +++++++++++++++++++
tests/unit.sh | 2 +-
13 files changed, 1483 insertions(+), 511 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/ab119806/tests/unit.sh
----------------------------------------------------------------------
[3/7] incubator-predictionio git commit: Add missing license
Posted by do...@apache.org.
Add missing license
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/bb003243
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/bb003243
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/bb003243
Branch: refs/heads/feature/xbuild
Commit: bb003243835ed20b6471af165303b9da17b09abf
Parents: 647b480
Author: Donald Szeto <do...@apache.org>
Authored: Wed Mar 15 11:31:28 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Wed Mar 15 11:31:28 2017 -0700
----------------------------------------------------------------------
docs/manual/source/templates/index.html.md | 25 +++++++++++++++++++++----
1 file changed, 21 insertions(+), 4 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/bb003243/docs/manual/source/templates/index.html.md
----------------------------------------------------------------------
diff --git a/docs/manual/source/templates/index.html.md b/docs/manual/source/templates/index.html.md
index 41b806b..f60699e 100644
--- a/docs/manual/source/templates/index.html.md
+++ b/docs/manual/source/templates/index.html.md
@@ -2,10 +2,27 @@
title: PredictionIO Official Templates
---
+<!--
+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
-PredictionIO's [template gallery](/gallery/template-gallery) offers both official and community contributed Engine Templates for all kinds of machine learning tasks.
-You can easily create one or more engines from these templates .
+ 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.
+-->
+
+PredictionIO's [template gallery](/gallery/template-gallery) offers both
+official and community contributed Engine Templates for all kinds of machine
+learning tasks. You can easily create one or more engines from these templates.
The section contains more detailed documentation of the official templates for
-beginner users to get familiar with PredictionIO and examples of how the templates
-can be modified for other usages.
+beginner users to get familiar with PredictionIO and examples of how the
+templates can be modified for other usages.
[2/7] incubator-predictionio git commit: Move storage depended tests
to each storage project
Posted by do...@apache.org.
Move storage depended tests to each storage project
Closes #362
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/647b480b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/647b480b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/647b480b
Branch: refs/heads/feature/xbuild
Commit: 647b480b0f0f693ec1b4d04dc9a2cf1492f0a0fe
Parents: 606be41
Author: Naoki Takezoe <ta...@gmail.com>
Authored: Wed Mar 15 09:42:30 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Wed Mar 15 09:42:30 2017 -0700
----------------------------------------------------------------------
.../predictionio/data/storage/LEventsSpec.scala | 248 -----------------
.../predictionio/data/storage/PEventsSpec.scala | 213 ---------------
.../data/storage/StorageTestUtils.scala | 45 ----
.../data/storage/hbase/LEventsSpec.scala | 239 +++++++++++++++++
.../data/storage/hbase/PEventsSpec.scala | 192 +++++++++++++
.../data/storage/hbase/StorageTestUtils.scala | 40 +++
.../data/storage/hbase/TestEvents.scala | 266 +++++++++++++++++++
.../data/storage/jdbc/LEventsSpec.scala | 236 ++++++++++++++++
.../data/storage/jdbc/PEventsSpec.scala | 193 ++++++++++++++
.../data/storage/jdbc/StorageTestUtils.scala | 29 ++
.../data/storage/jdbc/TestEvents.scala | 266 +++++++++++++++++++
11 files changed, 1461 insertions(+), 506 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/data/src/test/scala/org/apache/predictionio/data/storage/LEventsSpec.scala
----------------------------------------------------------------------
diff --git a/data/src/test/scala/org/apache/predictionio/data/storage/LEventsSpec.scala b/data/src/test/scala/org/apache/predictionio/data/storage/LEventsSpec.scala
deleted file mode 100644
index 3938072..0000000
--- a/data/src/test/scala/org/apache/predictionio/data/storage/LEventsSpec.scala
+++ /dev/null
@@ -1,248 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.predictionio.data.storage
-
-import org.specs2._
-import org.specs2.specification.Step
-
-class LEventsSpec extends Specification with TestEvents {
- def is = s2"""
-
- PredictionIO Storage LEvents Specification
-
- Events can be implemented by:
- - HBLEvents ${hbEvents}
- - JDBCLEvents ${jdbcLEvents}
-
- """
-
- def hbEvents = sequential ^ s2"""
-
- HBLEvents should
- - behave like any LEvents implementation ${events(hbDO)}
- - (table cleanup) ${Step(StorageTestUtils.dropHBaseNamespace(dbName))}
-
- """
-
- def jdbcLEvents = sequential ^ s2"""
-
- JDBCLEvents should
- - behave like any LEvents implementation ${events(jdbcDO)}
-
- """
-
- val appId = 1
-
- def events(eventClient: LEvents) = sequential ^ s2"""
-
- init default ${initDefault(eventClient)}
- insert 3 test events and get back by event ID ${insertAndGetEvents(eventClient)}
- insert 3 test events with timezone and get back by event ID ${insertAndGetTimezone(eventClient)}
- insert and delete by ID ${insertAndDelete(eventClient)}
- insert test user events ${insertTestUserEvents(eventClient)}
- find user events ${findUserEvents(eventClient)}
- aggregate user properties ${aggregateUserProperties(eventClient)}
- aggregate one user properties ${aggregateOneUserProperties(eventClient)}
- aggregate non-existent user properties ${aggregateNonExistentUserProperties(eventClient)}
- init channel ${initChannel(eventClient)}
- insert 2 events to channel ${insertChannel(eventClient)}
- insert 1 event to channel and delete by ID ${insertAndDeleteChannel(eventClient)}
- find events from channel ${findChannel(eventClient)}
- remove default ${removeDefault(eventClient)}
- remove channel ${removeChannel(eventClient)}
-
- """
-
- val dbName = "test_pio_storage_events_" + hashCode
- def hbDO = Storage.getDataObject[LEvents](
- StorageTestUtils.hbaseSourceName,
- dbName
- )
-
- def jdbcDO = Storage.getDataObject[LEvents](StorageTestUtils.jdbcSourceName, dbName)
-
- def initDefault(eventClient: LEvents) = {
- eventClient.init(appId)
- }
-
- def insertAndGetEvents(eventClient: LEvents) = {
-
- // events from TestEvents trait
- val listOfEvents = List(r1,r2,r3)
-
- val insertResp = listOfEvents.map { eventClient.insert(_, appId) }
-
- val insertedEventId: List[String] = insertResp
-
- val insertedEvent: List[Option[Event]] = listOfEvents.zip(insertedEventId)
- .map { case (e, id) => Some(e.copy(eventId = Some(id))) }
-
- val getResp = insertedEventId.map { id => eventClient.get(id, appId) }
-
- val getEvents = getResp
-
- insertedEvent must containTheSameElementsAs(getEvents)
- }
-
- def insertAndGetTimezone(eventClient: LEvents) = {
- val listOfEvents = List(tz1, tz2, tz3)
-
- val insertResp = listOfEvents.map { eventClient.insert(_, appId) }
-
- val insertedEventId: List[String] = insertResp
-
- val insertedEvent: List[Option[Event]] = listOfEvents.zip(insertedEventId)
- .map { case (e, id) => Some(e.copy(eventId = Some(id))) }
-
- val getResp = insertedEventId.map { id => eventClient.get(id, appId) }
-
- val getEvents = getResp
-
- insertedEvent must containTheSameElementsAs(getEvents)
- }
-
- def insertAndDelete(eventClient: LEvents) = {
- val eventId = eventClient.insert(r2, appId)
-
- val resultBefore = eventClient.get(eventId, appId)
-
- val expectedBefore = r2.copy(eventId = Some(eventId))
-
- val deleteStatus = eventClient.delete(eventId, appId)
-
- val resultAfter = eventClient.get(eventId, appId)
-
- (resultBefore must beEqualTo(Some(expectedBefore))) and
- (deleteStatus must beEqualTo(true)) and
- (resultAfter must beEqualTo(None))
- }
-
- def insertTestUserEvents(eventClient: LEvents) = {
- // events from TestEvents trait
- val listOfEvents = Vector(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2)
-
- listOfEvents.map{ eventClient.insert(_, appId) }
-
- success
- }
-
- def findUserEvents(eventClient: LEvents) = {
-
- val results: List[Event] = eventClient.find(
- appId = appId,
- entityType = Some("user"))
- .toList
- .map(e => e.copy(eventId = None)) // ignore eventID
-
- // same events in insertTestUserEvents
- val expected = List(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2)
-
- results must containTheSameElementsAs(expected)
- }
-
- def aggregateUserProperties(eventClient: LEvents) = {
-
- val result: Map[String, PropertyMap] = eventClient.aggregateProperties(
- appId = appId,
- entityType = "user")
-
- val expected = Map(
- "u1" -> PropertyMap(u1, u1BaseTime, u1LastTime),
- "u2" -> PropertyMap(u2, u2BaseTime, u2LastTime)
- )
-
- result must beEqualTo(expected)
- }
-
- def aggregateOneUserProperties(eventClient: LEvents) = {
- val result: Option[PropertyMap] = eventClient.aggregatePropertiesOfEntity(
- appId = appId,
- entityType = "user",
- entityId = "u1")
-
- val expected = Some(PropertyMap(u1, u1BaseTime, u1LastTime))
-
- result must beEqualTo(expected)
- }
-
- def aggregateNonExistentUserProperties(eventClient: LEvents) = {
- val result: Option[PropertyMap] = eventClient.aggregatePropertiesOfEntity(
- appId = appId,
- entityType = "user",
- entityId = "u999999")
-
- result must beEqualTo(None)
- }
-
- val channelId = 12
-
- def initChannel(eventClient: LEvents) = {
- eventClient.init(appId, Some(channelId))
- }
-
- def insertChannel(eventClient: LEvents) = {
-
- // events from TestEvents trait
- val listOfEvents = List(r4,r5)
-
- listOfEvents.map( eventClient.insert(_, appId, Some(channelId)) )
-
- success
- }
-
- def insertAndDeleteChannel(eventClient: LEvents) = {
-
- val eventId = eventClient.insert(r2, appId, Some(channelId))
-
- val resultBefore = eventClient.get(eventId, appId, Some(channelId))
-
- val expectedBefore = r2.copy(eventId = Some(eventId))
-
- val deleteStatus = eventClient.delete(eventId, appId, Some(channelId))
-
- val resultAfter = eventClient.get(eventId, appId, Some(channelId))
-
- (resultBefore must beEqualTo(Some(expectedBefore))) and
- (deleteStatus must beEqualTo(true)) and
- (resultAfter must beEqualTo(None))
- }
-
- def findChannel(eventClient: LEvents) = {
-
- val results: List[Event] = eventClient.find(
- appId = appId,
- channelId = Some(channelId)
- )
- .toList
- .map(e => e.copy(eventId = None)) // ignore eventId
-
- // same events in insertChannel
- val expected = List(r4, r5)
-
- results must containTheSameElementsAs(expected)
- }
-
- def removeDefault(eventClient: LEvents) = {
- eventClient.remove(appId)
- }
-
- def removeChannel(eventClient: LEvents) = {
- eventClient.remove(appId, Some(channelId))
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/data/src/test/scala/org/apache/predictionio/data/storage/PEventsSpec.scala
----------------------------------------------------------------------
diff --git a/data/src/test/scala/org/apache/predictionio/data/storage/PEventsSpec.scala b/data/src/test/scala/org/apache/predictionio/data/storage/PEventsSpec.scala
deleted file mode 100644
index ccd71a4..0000000
--- a/data/src/test/scala/org/apache/predictionio/data/storage/PEventsSpec.scala
+++ /dev/null
@@ -1,213 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.predictionio.data.storage
-
-import org.specs2._
-import org.specs2.specification.Step
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-class PEventsSpec extends Specification with TestEvents {
-
- System.clearProperty("spark.driver.port")
- System.clearProperty("spark.hostPort")
- val sc = new SparkContext("local[4]", "PEventAggregatorSpec test")
-
- val appId = 1
- val channelId = 6
- val dbName = "test_pio_storage_events_" + hashCode
-
- def hbLocal = Storage.getDataObject[LEvents](
- StorageTestUtils.hbaseSourceName,
- dbName
- )
-
- def hbPar = Storage.getDataObject[PEvents](
- StorageTestUtils.hbaseSourceName,
- dbName
- )
-
- def jdbcLocal = Storage.getDataObject[LEvents](
- StorageTestUtils.jdbcSourceName,
- dbName
- )
-
- def jdbcPar = Storage.getDataObject[PEvents](
- StorageTestUtils.jdbcSourceName,
- dbName
- )
-
- def stopSpark = {
- sc.stop()
- }
-
- def is = s2"""
-
- PredictionIO Storage PEvents Specification
-
- PEvents can be implemented by:
- - HBPEvents ${hbPEvents}
- - JDBCPEvents ${jdbcPEvents}
- - (stop Spark) ${Step(sc.stop())}
-
- """
-
- def hbPEvents = sequential ^ s2"""
-
- HBPEvents should
- - behave like any PEvents implementation ${events(hbLocal, hbPar)}
- - (table cleanup) ${Step(StorageTestUtils.dropHBaseNamespace(dbName))}
-
- """
-
- def jdbcPEvents = sequential ^ s2"""
-
- JDBCPEvents should
- - behave like any PEvents implementation ${events(jdbcLocal, jdbcPar)}
- - (table cleanup) ${Step(StorageTestUtils.dropJDBCTable(s"${dbName}_$appId"))}
- - (table cleanup) ${Step(StorageTestUtils.dropJDBCTable(s"${dbName}_${appId}_$channelId"))}
-
- """
-
- def events(localEventClient: LEvents, parEventClient: PEvents) = sequential ^ s2"""
-
- - (init test) ${initTest(localEventClient)}
- - (insert test events) ${insertTestEvents(localEventClient)}
- find in default ${find(parEventClient)}
- find in channel ${findChannel(parEventClient)}
- aggregate user properties in default ${aggregateUserProperties(parEventClient)}
- aggregate user properties in channel ${aggregateUserPropertiesChannel(parEventClient)}
- write to default ${write(parEventClient)}
- write to channel ${writeChannel(parEventClient)}
-
- """
-
- /* setup */
-
- // events from TestEvents trait
- val listOfEvents = List(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2, r1, r2)
- val listOfEventsChannel = List(u3e1, u3e2, u3e3, r3, r4)
-
- def initTest(localEventClient: LEvents) = {
- localEventClient.init(appId)
- localEventClient.init(appId, Some(channelId))
- }
-
- def insertTestEvents(localEventClient: LEvents) = {
- listOfEvents.map( localEventClient.insert(_, appId) )
- // insert to channel
- listOfEventsChannel.map( localEventClient.insert(_, appId, Some(channelId)) )
- success
- }
-
- /* following are tests */
-
- def find(parEventClient: PEvents) = {
- val resultRDD: RDD[Event] = parEventClient.find(
- appId = appId
- )(sc)
-
- val results = resultRDD.collect.toList
- .map {_.copy(eventId = None)} // ignore eventId
-
- results must containTheSameElementsAs(listOfEvents)
- }
-
- def findChannel(parEventClient: PEvents) = {
- val resultRDD: RDD[Event] = parEventClient.find(
- appId = appId,
- channelId = Some(channelId)
- )(sc)
-
- val results = resultRDD.collect.toList
- .map {_.copy(eventId = None)} // ignore eventId
-
- results must containTheSameElementsAs(listOfEventsChannel)
- }
-
- def aggregateUserProperties(parEventClient: PEvents) = {
- val resultRDD: RDD[(String, PropertyMap)] = parEventClient.aggregateProperties(
- appId = appId,
- entityType = "user"
- )(sc)
- val result: Map[String, PropertyMap] = resultRDD.collectAsMap.toMap
-
- val expected = Map(
- "u1" -> PropertyMap(u1, u1BaseTime, u1LastTime),
- "u2" -> PropertyMap(u2, u2BaseTime, u2LastTime)
- )
-
- result must beEqualTo(expected)
- }
-
- def aggregateUserPropertiesChannel(parEventClient: PEvents) = {
- val resultRDD: RDD[(String, PropertyMap)] = parEventClient.aggregateProperties(
- appId = appId,
- channelId = Some(channelId),
- entityType = "user"
- )(sc)
- val result: Map[String, PropertyMap] = resultRDD.collectAsMap.toMap
-
- val expected = Map(
- "u3" -> PropertyMap(u3, u3BaseTime, u3LastTime)
- )
-
- result must beEqualTo(expected)
- }
-
- def write(parEventClient: PEvents) = {
- val written = List(r5, r6)
- val writtenRDD = sc.parallelize(written)
- parEventClient.write(writtenRDD, appId)(sc)
-
- // read back
- val resultRDD = parEventClient.find(
- appId = appId
- )(sc)
-
- val results = resultRDD.collect.toList
- .map { _.copy(eventId = None)} // ignore eventId
-
- val expected = listOfEvents ++ written
-
- results must containTheSameElementsAs(expected)
- }
-
- def writeChannel(parEventClient: PEvents) = {
- val written = List(r1, r5, r6)
- val writtenRDD = sc.parallelize(written)
- parEventClient.write(writtenRDD, appId, Some(channelId))(sc)
-
- // read back
- val resultRDD = parEventClient.find(
- appId = appId,
- channelId = Some(channelId)
- )(sc)
-
- val results = resultRDD.collect.toList
- .map { _.copy(eventId = None)} // ignore eventId
-
- val expected = listOfEventsChannel ++ written
-
- results must containTheSameElementsAs(expected)
- }
-
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/data/src/test/scala/org/apache/predictionio/data/storage/StorageTestUtils.scala
----------------------------------------------------------------------
diff --git a/data/src/test/scala/org/apache/predictionio/data/storage/StorageTestUtils.scala b/data/src/test/scala/org/apache/predictionio/data/storage/StorageTestUtils.scala
deleted file mode 100644
index 747076a..0000000
--- a/data/src/test/scala/org/apache/predictionio/data/storage/StorageTestUtils.scala
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.predictionio.data.storage
-
-import org.apache.predictionio.data.storage.hbase.HBLEvents
-import scalikejdbc._
-
-object StorageTestUtils {
- val hbaseSourceName = "HBASE"
- val jdbcSourceName = "PGSQL"
-
- def dropHBaseNamespace(namespace: String): Unit = {
- val eventDb = Storage.getDataObject[LEvents](hbaseSourceName, namespace)
- .asInstanceOf[HBLEvents]
- val admin = eventDb.client.admin
- val tableNames = admin.listTableNamesByNamespace(namespace)
- tableNames.foreach { name =>
- admin.disableTable(name)
- admin.deleteTable(name)
- }
-
- //Only empty namespaces (no tables) can be removed.
- admin.deleteNamespace(namespace)
- }
-
- def dropJDBCTable(table: String): Unit = DB autoCommit { implicit s =>
- SQL(s"drop table $table").execute().apply()
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/LEventsSpec.scala
----------------------------------------------------------------------
diff --git a/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/LEventsSpec.scala b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/LEventsSpec.scala
new file mode 100644
index 0000000..c813ced
--- /dev/null
+++ b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/LEventsSpec.scala
@@ -0,0 +1,239 @@
+/*
+ * 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.predictionio.data.storage.hbase
+
+import org.apache.predictionio.data.storage.{Event, LEvents, PropertyMap, Storage}
+import org.specs2._
+import org.specs2.specification.Step
+
+class LEventsSpec extends Specification with TestEvents {
+ def is = s2"""
+
+ PredictionIO Storage LEvents Specification
+
+ Events can be implemented by:
+ - HBLEvents ${hbEvents}
+
+ """
+
+ def hbEvents = sequential ^ s2"""
+
+ HBLEvents should
+ - behave like any LEvents implementation ${events(hbDO)}
+ - (table cleanup) ${Step(StorageTestUtils.dropHBaseNamespace(dbName))}
+
+ """
+
+ val appId = 1
+
+ def events(eventClient: LEvents) = sequential ^ s2"""
+
+ init default ${initDefault(eventClient)}
+ insert 3 test events and get back by event ID ${insertAndGetEvents(eventClient)}
+ insert 3 test events with timezone and get back by event ID ${insertAndGetTimezone(eventClient)}
+ insert and delete by ID ${insertAndDelete(eventClient)}
+ insert test user events ${insertTestUserEvents(eventClient)}
+ find user events ${findUserEvents(eventClient)}
+ aggregate user properties ${aggregateUserProperties(eventClient)}
+ aggregate one user properties ${aggregateOneUserProperties(eventClient)}
+ aggregate non-existent user properties ${aggregateNonExistentUserProperties(eventClient)}
+ init channel ${initChannel(eventClient)}
+ insert 2 events to channel ${insertChannel(eventClient)}
+ insert 1 event to channel and delete by ID ${insertAndDeleteChannel(eventClient)}
+ find events from channel ${findChannel(eventClient)}
+ remove default ${removeDefault(eventClient)}
+ remove channel ${removeChannel(eventClient)}
+
+ """
+
+ val dbName = "test_pio_storage_events_" + hashCode
+ def hbDO = Storage.getDataObject[LEvents](
+ StorageTestUtils.hbaseSourceName,
+ dbName
+ )
+
+ def initDefault(eventClient: LEvents) = {
+ eventClient.init(appId)
+ }
+
+ def insertAndGetEvents(eventClient: LEvents) = {
+
+ // events from TestEvents trait
+ val listOfEvents = List(r1,r2,r3)
+
+ val insertResp = listOfEvents.map { eventClient.insert(_, appId) }
+
+ val insertedEventId: List[String] = insertResp
+
+ val insertedEvent: List[Option[Event]] = listOfEvents.zip(insertedEventId)
+ .map { case (e, id) => Some(e.copy(eventId = Some(id))) }
+
+ val getResp = insertedEventId.map { id => eventClient.get(id, appId) }
+
+ val getEvents = getResp
+
+ insertedEvent must containTheSameElementsAs(getEvents)
+ }
+
+ def insertAndGetTimezone(eventClient: LEvents) = {
+ val listOfEvents = List(tz1, tz2, tz3)
+
+ val insertResp = listOfEvents.map { eventClient.insert(_, appId) }
+
+ val insertedEventId: List[String] = insertResp
+
+ val insertedEvent: List[Option[Event]] = listOfEvents.zip(insertedEventId)
+ .map { case (e, id) => Some(e.copy(eventId = Some(id))) }
+
+ val getResp = insertedEventId.map { id => eventClient.get(id, appId) }
+
+ val getEvents = getResp
+
+ insertedEvent must containTheSameElementsAs(getEvents)
+ }
+
+ def insertAndDelete(eventClient: LEvents) = {
+ val eventId = eventClient.insert(r2, appId)
+
+ val resultBefore = eventClient.get(eventId, appId)
+
+ val expectedBefore = r2.copy(eventId = Some(eventId))
+
+ val deleteStatus = eventClient.delete(eventId, appId)
+
+ val resultAfter = eventClient.get(eventId, appId)
+
+ (resultBefore must beEqualTo(Some(expectedBefore))) and
+ (deleteStatus must beEqualTo(true)) and
+ (resultAfter must beEqualTo(None))
+ }
+
+ def insertTestUserEvents(eventClient: LEvents) = {
+ // events from TestEvents trait
+ val listOfEvents = Vector(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2)
+
+ listOfEvents.map{ eventClient.insert(_, appId) }
+
+ success
+ }
+
+ def findUserEvents(eventClient: LEvents) = {
+
+ val results: List[Event] = eventClient.find(
+ appId = appId,
+ entityType = Some("user"))
+ .toList
+ .map(e => e.copy(eventId = None)) // ignore eventID
+
+ // same events in insertTestUserEvents
+ val expected = List(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2)
+
+ results must containTheSameElementsAs(expected)
+ }
+
+ def aggregateUserProperties(eventClient: LEvents) = {
+
+ val result: Map[String, PropertyMap] = eventClient.aggregateProperties(
+ appId = appId,
+ entityType = "user")
+
+ val expected = Map(
+ "u1" -> PropertyMap(u1, u1BaseTime, u1LastTime),
+ "u2" -> PropertyMap(u2, u2BaseTime, u2LastTime)
+ )
+
+ result must beEqualTo(expected)
+ }
+
+ def aggregateOneUserProperties(eventClient: LEvents) = {
+ val result: Option[PropertyMap] = eventClient.aggregatePropertiesOfEntity(
+ appId = appId,
+ entityType = "user",
+ entityId = "u1")
+
+ val expected = Some(PropertyMap(u1, u1BaseTime, u1LastTime))
+
+ result must beEqualTo(expected)
+ }
+
+ def aggregateNonExistentUserProperties(eventClient: LEvents) = {
+ val result: Option[PropertyMap] = eventClient.aggregatePropertiesOfEntity(
+ appId = appId,
+ entityType = "user",
+ entityId = "u999999")
+
+ result must beEqualTo(None)
+ }
+
+ val channelId = 12
+
+ def initChannel(eventClient: LEvents) = {
+ eventClient.init(appId, Some(channelId))
+ }
+
+ def insertChannel(eventClient: LEvents) = {
+
+ // events from TestEvents trait
+ val listOfEvents = List(r4,r5)
+
+ listOfEvents.map( eventClient.insert(_, appId, Some(channelId)) )
+
+ success
+ }
+
+ def insertAndDeleteChannel(eventClient: LEvents) = {
+
+ val eventId = eventClient.insert(r2, appId, Some(channelId))
+
+ val resultBefore = eventClient.get(eventId, appId, Some(channelId))
+
+ val expectedBefore = r2.copy(eventId = Some(eventId))
+
+ val deleteStatus = eventClient.delete(eventId, appId, Some(channelId))
+
+ val resultAfter = eventClient.get(eventId, appId, Some(channelId))
+
+ (resultBefore must beEqualTo(Some(expectedBefore))) and
+ (deleteStatus must beEqualTo(true)) and
+ (resultAfter must beEqualTo(None))
+ }
+
+ def findChannel(eventClient: LEvents) = {
+
+ val results: List[Event] = eventClient.find(
+ appId = appId,
+ channelId = Some(channelId)
+ )
+ .toList
+ .map(e => e.copy(eventId = None)) // ignore eventId
+
+ // same events in insertChannel
+ val expected = List(r4, r5)
+
+ results must containTheSameElementsAs(expected)
+ }
+
+ def removeDefault(eventClient: LEvents) = {
+ eventClient.remove(appId)
+ }
+
+ def removeChannel(eventClient: LEvents) = {
+ eventClient.remove(appId, Some(channelId))
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/PEventsSpec.scala
----------------------------------------------------------------------
diff --git a/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/PEventsSpec.scala b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/PEventsSpec.scala
new file mode 100644
index 0000000..d675e55
--- /dev/null
+++ b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/PEventsSpec.scala
@@ -0,0 +1,192 @@
+/*
+ * 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.predictionio.data.storage.hbase
+
+import org.apache.predictionio.data.storage._
+import org.apache.spark.SparkContext
+import org.apache.spark.rdd.RDD
+import org.specs2._
+import org.specs2.specification.Step
+
+class PEventsSpec extends Specification with TestEvents {
+
+ System.clearProperty("spark.driver.port")
+ System.clearProperty("spark.hostPort")
+ val sc = new SparkContext("local[4]", "PEventAggregatorSpec test")
+
+ val appId = 1
+ val channelId = 6
+ val dbName = "test_pio_storage_events_" + hashCode
+
+ def hbLocal = Storage.getDataObject[LEvents](
+ StorageTestUtils.hbaseSourceName,
+ dbName
+ )
+
+ def hbPar = Storage.getDataObject[PEvents](
+ StorageTestUtils.hbaseSourceName,
+ dbName
+ )
+
+ def stopSpark = {
+ sc.stop()
+ }
+
+ def is = s2"""
+
+ PredictionIO Storage PEvents Specification
+
+ PEvents can be implemented by:
+ - HBPEvents ${hbPEvents}
+ - (stop Spark) ${Step(sc.stop())}
+
+ """
+
+ def hbPEvents = sequential ^ s2"""
+
+ HBPEvents should
+ - behave like any PEvents implementation ${events(hbLocal, hbPar)}
+ - (table cleanup) ${Step(StorageTestUtils.dropHBaseNamespace(dbName))}
+
+ """
+
+ def events(localEventClient: LEvents, parEventClient: PEvents) = sequential ^ s2"""
+
+ - (init test) ${initTest(localEventClient)}
+ - (insert test events) ${insertTestEvents(localEventClient)}
+ find in default ${find(parEventClient)}
+ find in channel ${findChannel(parEventClient)}
+ aggregate user properties in default ${aggregateUserProperties(parEventClient)}
+ aggregate user properties in channel ${aggregateUserPropertiesChannel(parEventClient)}
+ write to default ${write(parEventClient)}
+ write to channel ${writeChannel(parEventClient)}
+
+ """
+
+ /* setup */
+
+ // events from TestEvents trait
+ val listOfEvents = List(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2, r1, r2)
+ val listOfEventsChannel = List(u3e1, u3e2, u3e3, r3, r4)
+
+ def initTest(localEventClient: LEvents) = {
+ localEventClient.init(appId)
+ localEventClient.init(appId, Some(channelId))
+ }
+
+ def insertTestEvents(localEventClient: LEvents) = {
+ listOfEvents.map( localEventClient.insert(_, appId) )
+ // insert to channel
+ listOfEventsChannel.map( localEventClient.insert(_, appId, Some(channelId)) )
+ success
+ }
+
+ /* following are tests */
+
+ def find(parEventClient: PEvents) = {
+ val resultRDD: RDD[Event] = parEventClient.find(
+ appId = appId
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map {_.copy(eventId = None)} // ignore eventId
+
+ results must containTheSameElementsAs(listOfEvents)
+ }
+
+ def findChannel(parEventClient: PEvents) = {
+ val resultRDD: RDD[Event] = parEventClient.find(
+ appId = appId,
+ channelId = Some(channelId)
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map {_.copy(eventId = None)} // ignore eventId
+
+ results must containTheSameElementsAs(listOfEventsChannel)
+ }
+
+ def aggregateUserProperties(parEventClient: PEvents) = {
+ val resultRDD: RDD[(String, PropertyMap)] = parEventClient.aggregateProperties(
+ appId = appId,
+ entityType = "user"
+ )(sc)
+ val result: Map[String, PropertyMap] = resultRDD.collectAsMap.toMap
+
+ val expected = Map(
+ "u1" -> PropertyMap(u1, u1BaseTime, u1LastTime),
+ "u2" -> PropertyMap(u2, u2BaseTime, u2LastTime)
+ )
+
+ result must beEqualTo(expected)
+ }
+
+ def aggregateUserPropertiesChannel(parEventClient: PEvents) = {
+ val resultRDD: RDD[(String, PropertyMap)] = parEventClient.aggregateProperties(
+ appId = appId,
+ channelId = Some(channelId),
+ entityType = "user"
+ )(sc)
+ val result: Map[String, PropertyMap] = resultRDD.collectAsMap.toMap
+
+ val expected = Map(
+ "u3" -> PropertyMap(u3, u3BaseTime, u3LastTime)
+ )
+
+ result must beEqualTo(expected)
+ }
+
+ def write(parEventClient: PEvents) = {
+ val written = List(r5, r6)
+ val writtenRDD = sc.parallelize(written)
+ parEventClient.write(writtenRDD, appId)(sc)
+
+ // read back
+ val resultRDD = parEventClient.find(
+ appId = appId
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map { _.copy(eventId = None)} // ignore eventId
+
+ val expected = listOfEvents ++ written
+
+ results must containTheSameElementsAs(expected)
+ }
+
+ def writeChannel(parEventClient: PEvents) = {
+ val written = List(r1, r5, r6)
+ val writtenRDD = sc.parallelize(written)
+ parEventClient.write(writtenRDD, appId, Some(channelId))(sc)
+
+ // read back
+ val resultRDD = parEventClient.find(
+ appId = appId,
+ channelId = Some(channelId)
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map { _.copy(eventId = None)} // ignore eventId
+
+ val expected = listOfEventsChannel ++ written
+
+ results must containTheSameElementsAs(expected)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/StorageTestUtils.scala
----------------------------------------------------------------------
diff --git a/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/StorageTestUtils.scala b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/StorageTestUtils.scala
new file mode 100644
index 0000000..161cf90
--- /dev/null
+++ b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/StorageTestUtils.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.predictionio.data.storage.hbase
+
+import org.apache.predictionio.data.storage.{LEvents, Storage}
+
+object StorageTestUtils {
+ val hbaseSourceName = "HBASE"
+
+ def dropHBaseNamespace(namespace: String): Unit = {
+ val eventDb = Storage.getDataObject[LEvents](hbaseSourceName, namespace)
+ .asInstanceOf[HBLEvents]
+ val admin = eventDb.client.admin
+ val tableNames = admin.listTableNamesByNamespace(namespace)
+ tableNames.foreach { name =>
+ admin.disableTable(name)
+ admin.deleteTable(name)
+ }
+
+ // Only empty namespaces (no tables) can be removed.
+ admin.deleteNamespace(namespace)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/TestEvents.scala
----------------------------------------------------------------------
diff --git a/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/TestEvents.scala b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/TestEvents.scala
new file mode 100644
index 0000000..2171864
--- /dev/null
+++ b/storage/hbase/src/test/scala/org/apache/predictionio/data/storage/hbase/TestEvents.scala
@@ -0,0 +1,266 @@
+/*
+ * 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.predictionio.data.storage.hbase
+
+import org.apache.predictionio.data.storage.{DataMap, Event}
+import org.joda.time.{DateTime, DateTimeZone}
+
+trait TestEvents {
+
+ val u1BaseTime = new DateTime(654321)
+ val u2BaseTime = new DateTime(6543210)
+ val u3BaseTime = new DateTime(6543410)
+
+ // u1 events
+ val u1e1 = Event(
+ event = "$set",
+ entityType = "user",
+ entityId = "u1",
+ properties = DataMap(
+ """{
+ "a" : 1,
+ "b" : "value2",
+ "d" : [1, 2, 3],
+ }"""),
+ eventTime = u1BaseTime
+ )
+
+ val u1e2 = u1e1.copy(
+ event = "$set",
+ properties = DataMap("""{"a" : 2}"""),
+ eventTime = u1BaseTime.plusDays(1)
+ )
+
+ val u1e3 = u1e1.copy(
+ event = "$set",
+ properties = DataMap("""{"b" : "value4"}"""),
+ eventTime = u1BaseTime.plusDays(2)
+ )
+
+ val u1e4 = u1e1.copy(
+ event = "$unset",
+ properties = DataMap("""{"b" : null}"""),
+ eventTime = u1BaseTime.plusDays(3)
+ )
+
+ val u1e5 = u1e1.copy(
+ event = "$set",
+ properties = DataMap("""{"e" : "new"}"""),
+ eventTime = u1BaseTime.plusDays(4)
+ )
+
+ val u1LastTime = u1BaseTime.plusDays(4)
+ val u1 = """{"a": 2, "d": [1, 2, 3], "e": "new"}"""
+
+ // delete event for u1
+ val u1ed = u1e1.copy(
+ event = "$delete",
+ properties = DataMap(),
+ eventTime = u1BaseTime.plusDays(5)
+ )
+
+ // u2 events
+ val u2e1 = Event(
+ event = "$set",
+ entityType = "user",
+ entityId = "u2",
+ properties = DataMap(
+ """{
+ "a" : 21,
+ "b" : "value12",
+ "d" : [7, 5, 6],
+ }"""),
+ eventTime = u2BaseTime
+ )
+
+ val u2e2 = u2e1.copy(
+ event = "$unset",
+ properties = DataMap("""{"a" : null}"""),
+ eventTime = u2BaseTime.plusDays(1)
+ )
+
+ val u2e3 = u2e1.copy(
+ event = "$set",
+ properties = DataMap("""{"b" : "value9", "g": "new11"}"""),
+ eventTime = u2BaseTime.plusDays(2)
+ )
+
+ val u2LastTime = u2BaseTime.plusDays(2)
+ val u2 = """{"b": "value9", "d": [7, 5, 6], "g": "new11"}"""
+
+ // u3 events
+ val u3e1 = Event(
+ event = "$set",
+ entityType = "user",
+ entityId = "u3",
+ properties = DataMap(
+ """{
+ "a" : 22,
+ "b" : "value13",
+ "d" : [5, 6, 1],
+ }"""),
+ eventTime = u3BaseTime
+ )
+
+ val u3e2 = u3e1.copy(
+ event = "$unset",
+ properties = DataMap("""{"a" : null}"""),
+ eventTime = u3BaseTime.plusDays(1)
+ )
+
+ val u3e3 = u3e1.copy(
+ event = "$set",
+ properties = DataMap("""{"b" : "value10", "f": "new12", "d" : [1, 3, 2]}"""),
+ eventTime = u3BaseTime.plusDays(2)
+ )
+
+ val u3LastTime = u3BaseTime.plusDays(2)
+ val u3 = """{"b": "value10", "d": [1, 3, 2], "f": "new12"}"""
+
+ // some random events
+ val r1 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id",
+ targetEntityType = Some("my_target_entity_type"),
+ targetEntityId = Some("my_target_entity_id"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = DateTime.now,
+ prId = Some("my_prid")
+ )
+ val r2 = Event(
+ event = "my_event2",
+ entityType = "my_entity_type2",
+ entityId = "my_entity_id2"
+ )
+ val r3 = Event(
+ event = "my_event3",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id",
+ targetEntityType = Some("my_target_entity_type"),
+ targetEntityId = Some("my_target_entity_id"),
+ properties = DataMap(
+ """{
+ "propA" : 1.2345,
+ "propB" : "valueB",
+ }"""
+ ),
+ prId = Some("my_prid")
+ )
+ val r4 = Event(
+ event = "my_event4",
+ entityType = "my_entity_type4",
+ entityId = "my_entity_id4",
+ targetEntityType = Some("my_target_entity_type4"),
+ targetEntityId = Some("my_target_entity_id4"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""),
+ eventTime = DateTime.now
+ )
+ val r5 = Event(
+ event = "my_event5",
+ entityType = "my_entity_type5",
+ entityId = "my_entity_id5",
+ targetEntityType = Some("my_target_entity_type5"),
+ targetEntityId = Some("my_target_entity_id5"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = DateTime.now
+ )
+ val r6 = Event(
+ event = "my_event6",
+ entityType = "my_entity_type6",
+ entityId = "my_entity_id6",
+ targetEntityType = Some("my_target_entity_type6"),
+ targetEntityId = Some("my_target_entity_id6"),
+ properties = DataMap(
+ """{
+ "prop1" : 6,
+ "prop2" : "value2",
+ "prop3" : [6, 7, 8],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = DateTime.now
+ )
+
+ // timezone
+ val tz1 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id0",
+ targetEntityType = Some("my_target_entity_type"),
+ targetEntityId = Some("my_target_entity_id"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = new DateTime(12345678, DateTimeZone.forID("-08:00")),
+ prId = Some("my_prid")
+ )
+
+ val tz2 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id1",
+ eventTime = new DateTime(12345678, DateTimeZone.forID("+02:00")),
+ prId = Some("my_prid")
+ )
+
+ val tz3 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id2",
+ eventTime = new DateTime(12345678, DateTimeZone.forID("+08:00")),
+ prId = Some("my_prid")
+ )
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/LEventsSpec.scala
----------------------------------------------------------------------
diff --git a/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/LEventsSpec.scala b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/LEventsSpec.scala
new file mode 100644
index 0000000..d723d07
--- /dev/null
+++ b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/LEventsSpec.scala
@@ -0,0 +1,236 @@
+/*
+ * 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.predictionio.data.storage.jdbc
+
+import org.apache.predictionio.data.storage.{Event, LEvents, PropertyMap, Storage}
+import org.specs2._
+import org.specs2.specification.Step
+
+class LEventsSpec extends Specification with TestEvents {
+ def is = s2"""
+
+ PredictionIO Storage LEvents Specification
+
+ Events can be implemented by:
+ - JDBCLEvents ${jdbcLEvents}
+
+ """
+
+ def jdbcLEvents = sequential ^ s2"""
+
+ JDBCLEvents should
+ - behave like any LEvents implementation ${events(jdbcDO)}
+
+ """
+
+ val appId = 1
+
+ def events(eventClient: LEvents) = sequential ^ s2"""
+
+ init default ${initDefault(eventClient)}
+ insert 3 test events and get back by event ID ${insertAndGetEvents(eventClient)}
+ insert 3 test events with timezone and get back by event ID ${insertAndGetTimezone(eventClient)}
+ insert and delete by ID ${insertAndDelete(eventClient)}
+ insert test user events ${insertTestUserEvents(eventClient)}
+ find user events ${findUserEvents(eventClient)}
+ aggregate user properties ${aggregateUserProperties(eventClient)}
+ aggregate one user properties ${aggregateOneUserProperties(eventClient)}
+ aggregate non-existent user properties ${aggregateNonExistentUserProperties(eventClient)}
+ init channel ${initChannel(eventClient)}
+ insert 2 events to channel ${insertChannel(eventClient)}
+ insert 1 event to channel and delete by ID ${insertAndDeleteChannel(eventClient)}
+ find events from channel ${findChannel(eventClient)}
+ remove default ${removeDefault(eventClient)}
+ remove channel ${removeChannel(eventClient)}
+
+ """
+
+ val dbName = "test_pio_storage_events_" + hashCode
+
+ def jdbcDO = Storage.getDataObject[LEvents](StorageTestUtils.jdbcSourceName, dbName)
+
+ def initDefault(eventClient: LEvents) = {
+ eventClient.init(appId)
+ }
+
+ def insertAndGetEvents(eventClient: LEvents) = {
+
+ // events from TestEvents trait
+ val listOfEvents = List(r1,r2,r3)
+
+ val insertResp = listOfEvents.map { eventClient.insert(_, appId) }
+
+ val insertedEventId: List[String] = insertResp
+
+ val insertedEvent: List[Option[Event]] = listOfEvents.zip(insertedEventId)
+ .map { case (e, id) => Some(e.copy(eventId = Some(id))) }
+
+ val getResp = insertedEventId.map { id => eventClient.get(id, appId) }
+
+ val getEvents = getResp
+
+ insertedEvent must containTheSameElementsAs(getEvents)
+ }
+
+ def insertAndGetTimezone(eventClient: LEvents) = {
+ val listOfEvents = List(tz1, tz2, tz3)
+
+ val insertResp = listOfEvents.map { eventClient.insert(_, appId) }
+
+ val insertedEventId: List[String] = insertResp
+
+ val insertedEvent: List[Option[Event]] = listOfEvents.zip(insertedEventId)
+ .map { case (e, id) => Some(e.copy(eventId = Some(id))) }
+
+ val getResp = insertedEventId.map { id => eventClient.get(id, appId) }
+
+ val getEvents = getResp
+
+ insertedEvent must containTheSameElementsAs(getEvents)
+ }
+
+ def insertAndDelete(eventClient: LEvents) = {
+ val eventId = eventClient.insert(r2, appId)
+
+ val resultBefore = eventClient.get(eventId, appId)
+
+ val expectedBefore = r2.copy(eventId = Some(eventId))
+
+ val deleteStatus = eventClient.delete(eventId, appId)
+
+ val resultAfter = eventClient.get(eventId, appId)
+
+ (resultBefore must beEqualTo(Some(expectedBefore))) and
+ (deleteStatus must beEqualTo(true)) and
+ (resultAfter must beEqualTo(None))
+ }
+
+ def insertTestUserEvents(eventClient: LEvents) = {
+ // events from TestEvents trait
+ val listOfEvents = Vector(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2)
+
+ listOfEvents.map{ eventClient.insert(_, appId) }
+
+ success
+ }
+
+ def findUserEvents(eventClient: LEvents) = {
+
+ val results: List[Event] = eventClient.find(
+ appId = appId,
+ entityType = Some("user"))
+ .toList
+ .map(e => e.copy(eventId = None)) // ignore eventID
+
+ // same events in insertTestUserEvents
+ val expected = List(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2)
+
+ results must containTheSameElementsAs(expected)
+ }
+
+ def aggregateUserProperties(eventClient: LEvents) = {
+
+ val result: Map[String, PropertyMap] = eventClient.aggregateProperties(
+ appId = appId,
+ entityType = "user")
+
+ val expected = Map(
+ "u1" -> PropertyMap(u1, u1BaseTime, u1LastTime),
+ "u2" -> PropertyMap(u2, u2BaseTime, u2LastTime)
+ )
+
+ result must beEqualTo(expected)
+ }
+
+ def aggregateOneUserProperties(eventClient: LEvents) = {
+ val result: Option[PropertyMap] = eventClient.aggregatePropertiesOfEntity(
+ appId = appId,
+ entityType = "user",
+ entityId = "u1")
+
+ val expected = Some(PropertyMap(u1, u1BaseTime, u1LastTime))
+
+ result must beEqualTo(expected)
+ }
+
+ def aggregateNonExistentUserProperties(eventClient: LEvents) = {
+ val result: Option[PropertyMap] = eventClient.aggregatePropertiesOfEntity(
+ appId = appId,
+ entityType = "user",
+ entityId = "u999999")
+
+ result must beEqualTo(None)
+ }
+
+ val channelId = 12
+
+ def initChannel(eventClient: LEvents) = {
+ eventClient.init(appId, Some(channelId))
+ }
+
+ def insertChannel(eventClient: LEvents) = {
+
+ // events from TestEvents trait
+ val listOfEvents = List(r4,r5)
+
+ listOfEvents.map( eventClient.insert(_, appId, Some(channelId)) )
+
+ success
+ }
+
+ def insertAndDeleteChannel(eventClient: LEvents) = {
+
+ val eventId = eventClient.insert(r2, appId, Some(channelId))
+
+ val resultBefore = eventClient.get(eventId, appId, Some(channelId))
+
+ val expectedBefore = r2.copy(eventId = Some(eventId))
+
+ val deleteStatus = eventClient.delete(eventId, appId, Some(channelId))
+
+ val resultAfter = eventClient.get(eventId, appId, Some(channelId))
+
+ (resultBefore must beEqualTo(Some(expectedBefore))) and
+ (deleteStatus must beEqualTo(true)) and
+ (resultAfter must beEqualTo(None))
+ }
+
+ def findChannel(eventClient: LEvents) = {
+
+ val results: List[Event] = eventClient.find(
+ appId = appId,
+ channelId = Some(channelId)
+ )
+ .toList
+ .map(e => e.copy(eventId = None)) // ignore eventId
+
+ // same events in insertChannel
+ val expected = List(r4, r5)
+
+ results must containTheSameElementsAs(expected)
+ }
+
+ def removeDefault(eventClient: LEvents) = {
+ eventClient.remove(appId)
+ }
+
+ def removeChannel(eventClient: LEvents) = {
+ eventClient.remove(appId, Some(channelId))
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/PEventsSpec.scala
----------------------------------------------------------------------
diff --git a/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/PEventsSpec.scala b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/PEventsSpec.scala
new file mode 100644
index 0000000..71ebf5f
--- /dev/null
+++ b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/PEventsSpec.scala
@@ -0,0 +1,193 @@
+/*
+ * 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.predictionio.data.storage.jdbc
+
+import org.apache.predictionio.data.storage._
+import org.apache.spark.SparkContext
+import org.apache.spark.rdd.RDD
+import org.specs2._
+import org.specs2.specification.Step
+
+class PEventsSpec extends Specification with TestEvents {
+
+ System.clearProperty("spark.driver.port")
+ System.clearProperty("spark.hostPort")
+ val sc = new SparkContext("local[4]", "PEventAggregatorSpec test")
+
+ val appId = 1
+ val channelId = 6
+ val dbName = "test_pio_storage_events_" + hashCode
+
+ def jdbcLocal = Storage.getDataObject[LEvents](
+ StorageTestUtils.jdbcSourceName,
+ dbName
+ )
+
+ def jdbcPar = Storage.getDataObject[PEvents](
+ StorageTestUtils.jdbcSourceName,
+ dbName
+ )
+
+ def stopSpark = {
+ sc.stop()
+ }
+
+ def is = s2"""
+
+ PredictionIO Storage PEvents Specification
+
+ PEvents can be implemented by:
+ - JDBCPEvents ${jdbcPEvents}
+ - (stop Spark) ${Step(sc.stop())}
+
+ """
+
+ def jdbcPEvents = sequential ^ s2"""
+
+ JDBCPEvents should
+ - behave like any PEvents implementation ${events(jdbcLocal, jdbcPar)}
+ - (table cleanup) ${Step(StorageTestUtils.dropJDBCTable(s"${dbName}_$appId"))}
+ - (table cleanup) ${Step(StorageTestUtils.dropJDBCTable(s"${dbName}_${appId}_$channelId"))}
+
+ """
+
+ def events(localEventClient: LEvents, parEventClient: PEvents) = sequential ^ s2"""
+
+ - (init test) ${initTest(localEventClient)}
+ - (insert test events) ${insertTestEvents(localEventClient)}
+ find in default ${find(parEventClient)}
+ find in channel ${findChannel(parEventClient)}
+ aggregate user properties in default ${aggregateUserProperties(parEventClient)}
+ aggregate user properties in channel ${aggregateUserPropertiesChannel(parEventClient)}
+ write to default ${write(parEventClient)}
+ write to channel ${writeChannel(parEventClient)}
+
+ """
+
+ /* setup */
+
+ // events from TestEvents trait
+ val listOfEvents = List(u1e5, u2e2, u1e3, u1e1, u2e3, u2e1, u1e4, u1e2, r1, r2)
+ val listOfEventsChannel = List(u3e1, u3e2, u3e3, r3, r4)
+
+ def initTest(localEventClient: LEvents) = {
+ localEventClient.init(appId)
+ localEventClient.init(appId, Some(channelId))
+ }
+
+ def insertTestEvents(localEventClient: LEvents) = {
+ listOfEvents.map( localEventClient.insert(_, appId) )
+ // insert to channel
+ listOfEventsChannel.map( localEventClient.insert(_, appId, Some(channelId)) )
+ success
+ }
+
+ /* following are tests */
+
+ def find(parEventClient: PEvents) = {
+ val resultRDD: RDD[Event] = parEventClient.find(
+ appId = appId
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map {_.copy(eventId = None)} // ignore eventId
+
+ results must containTheSameElementsAs(listOfEvents)
+ }
+
+ def findChannel(parEventClient: PEvents) = {
+ val resultRDD: RDD[Event] = parEventClient.find(
+ appId = appId,
+ channelId = Some(channelId)
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map {_.copy(eventId = None)} // ignore eventId
+
+ results must containTheSameElementsAs(listOfEventsChannel)
+ }
+
+ def aggregateUserProperties(parEventClient: PEvents) = {
+ val resultRDD: RDD[(String, PropertyMap)] = parEventClient.aggregateProperties(
+ appId = appId,
+ entityType = "user"
+ )(sc)
+ val result: Map[String, PropertyMap] = resultRDD.collectAsMap.toMap
+
+ val expected = Map(
+ "u1" -> PropertyMap(u1, u1BaseTime, u1LastTime),
+ "u2" -> PropertyMap(u2, u2BaseTime, u2LastTime)
+ )
+
+ result must beEqualTo(expected)
+ }
+
+ def aggregateUserPropertiesChannel(parEventClient: PEvents) = {
+ val resultRDD: RDD[(String, PropertyMap)] = parEventClient.aggregateProperties(
+ appId = appId,
+ channelId = Some(channelId),
+ entityType = "user"
+ )(sc)
+ val result: Map[String, PropertyMap] = resultRDD.collectAsMap.toMap
+
+ val expected = Map(
+ "u3" -> PropertyMap(u3, u3BaseTime, u3LastTime)
+ )
+
+ result must beEqualTo(expected)
+ }
+
+ def write(parEventClient: PEvents) = {
+ val written = List(r5, r6)
+ val writtenRDD = sc.parallelize(written)
+ parEventClient.write(writtenRDD, appId)(sc)
+
+ // read back
+ val resultRDD = parEventClient.find(
+ appId = appId
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map { _.copy(eventId = None)} // ignore eventId
+
+ val expected = listOfEvents ++ written
+
+ results must containTheSameElementsAs(expected)
+ }
+
+ def writeChannel(parEventClient: PEvents) = {
+ val written = List(r1, r5, r6)
+ val writtenRDD = sc.parallelize(written)
+ parEventClient.write(writtenRDD, appId, Some(channelId))(sc)
+
+ // read back
+ val resultRDD = parEventClient.find(
+ appId = appId,
+ channelId = Some(channelId)
+ )(sc)
+
+ val results = resultRDD.collect.toList
+ .map { _.copy(eventId = None)} // ignore eventId
+
+ val expected = listOfEventsChannel ++ written
+
+ results must containTheSameElementsAs(expected)
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/StorageTestUtils.scala
----------------------------------------------------------------------
diff --git a/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/StorageTestUtils.scala b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/StorageTestUtils.scala
new file mode 100644
index 0000000..4bf90cf
--- /dev/null
+++ b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/StorageTestUtils.scala
@@ -0,0 +1,29 @@
+/*
+ * 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.predictionio.data.storage.jdbc
+
+import scalikejdbc._
+
+object StorageTestUtils {
+ val jdbcSourceName = "PGSQL"
+
+ def dropJDBCTable(table: String): Unit = DB autoCommit { implicit s =>
+ SQL(s"drop table $table").execute().apply()
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/647b480b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/TestEvents.scala
----------------------------------------------------------------------
diff --git a/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/TestEvents.scala b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/TestEvents.scala
new file mode 100644
index 0000000..2cb08e5
--- /dev/null
+++ b/storage/jdbc/src/test/scala/org/apache/predictionio/data/storage/jdbc/TestEvents.scala
@@ -0,0 +1,266 @@
+/*
+ * 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.predictionio.data.storage.jdbc
+
+import org.apache.predictionio.data.storage.{DataMap, Event}
+import org.joda.time.{DateTime, DateTimeZone}
+
+trait TestEvents {
+
+ val u1BaseTime = new DateTime(654321)
+ val u2BaseTime = new DateTime(6543210)
+ val u3BaseTime = new DateTime(6543410)
+
+ // u1 events
+ val u1e1 = Event(
+ event = "$set",
+ entityType = "user",
+ entityId = "u1",
+ properties = DataMap(
+ """{
+ "a" : 1,
+ "b" : "value2",
+ "d" : [1, 2, 3],
+ }"""),
+ eventTime = u1BaseTime
+ )
+
+ val u1e2 = u1e1.copy(
+ event = "$set",
+ properties = DataMap("""{"a" : 2}"""),
+ eventTime = u1BaseTime.plusDays(1)
+ )
+
+ val u1e3 = u1e1.copy(
+ event = "$set",
+ properties = DataMap("""{"b" : "value4"}"""),
+ eventTime = u1BaseTime.plusDays(2)
+ )
+
+ val u1e4 = u1e1.copy(
+ event = "$unset",
+ properties = DataMap("""{"b" : null}"""),
+ eventTime = u1BaseTime.plusDays(3)
+ )
+
+ val u1e5 = u1e1.copy(
+ event = "$set",
+ properties = DataMap("""{"e" : "new"}"""),
+ eventTime = u1BaseTime.plusDays(4)
+ )
+
+ val u1LastTime = u1BaseTime.plusDays(4)
+ val u1 = """{"a": 2, "d": [1, 2, 3], "e": "new"}"""
+
+ // delete event for u1
+ val u1ed = u1e1.copy(
+ event = "$delete",
+ properties = DataMap(),
+ eventTime = u1BaseTime.plusDays(5)
+ )
+
+ // u2 events
+ val u2e1 = Event(
+ event = "$set",
+ entityType = "user",
+ entityId = "u2",
+ properties = DataMap(
+ """{
+ "a" : 21,
+ "b" : "value12",
+ "d" : [7, 5, 6],
+ }"""),
+ eventTime = u2BaseTime
+ )
+
+ val u2e2 = u2e1.copy(
+ event = "$unset",
+ properties = DataMap("""{"a" : null}"""),
+ eventTime = u2BaseTime.plusDays(1)
+ )
+
+ val u2e3 = u2e1.copy(
+ event = "$set",
+ properties = DataMap("""{"b" : "value9", "g": "new11"}"""),
+ eventTime = u2BaseTime.plusDays(2)
+ )
+
+ val u2LastTime = u2BaseTime.plusDays(2)
+ val u2 = """{"b": "value9", "d": [7, 5, 6], "g": "new11"}"""
+
+ // u3 events
+ val u3e1 = Event(
+ event = "$set",
+ entityType = "user",
+ entityId = "u3",
+ properties = DataMap(
+ """{
+ "a" : 22,
+ "b" : "value13",
+ "d" : [5, 6, 1],
+ }"""),
+ eventTime = u3BaseTime
+ )
+
+ val u3e2 = u3e1.copy(
+ event = "$unset",
+ properties = DataMap("""{"a" : null}"""),
+ eventTime = u3BaseTime.plusDays(1)
+ )
+
+ val u3e3 = u3e1.copy(
+ event = "$set",
+ properties = DataMap("""{"b" : "value10", "f": "new12", "d" : [1, 3, 2]}"""),
+ eventTime = u3BaseTime.plusDays(2)
+ )
+
+ val u3LastTime = u3BaseTime.plusDays(2)
+ val u3 = """{"b": "value10", "d": [1, 3, 2], "f": "new12"}"""
+
+ // some random events
+ val r1 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id",
+ targetEntityType = Some("my_target_entity_type"),
+ targetEntityId = Some("my_target_entity_id"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = DateTime.now,
+ prId = Some("my_prid")
+ )
+ val r2 = Event(
+ event = "my_event2",
+ entityType = "my_entity_type2",
+ entityId = "my_entity_id2"
+ )
+ val r3 = Event(
+ event = "my_event3",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id",
+ targetEntityType = Some("my_target_entity_type"),
+ targetEntityId = Some("my_target_entity_id"),
+ properties = DataMap(
+ """{
+ "propA" : 1.2345,
+ "propB" : "valueB",
+ }"""
+ ),
+ prId = Some("my_prid")
+ )
+ val r4 = Event(
+ event = "my_event4",
+ entityType = "my_entity_type4",
+ entityId = "my_entity_id4",
+ targetEntityType = Some("my_target_entity_type4"),
+ targetEntityId = Some("my_target_entity_id4"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""),
+ eventTime = DateTime.now
+ )
+ val r5 = Event(
+ event = "my_event5",
+ entityType = "my_entity_type5",
+ entityId = "my_entity_id5",
+ targetEntityType = Some("my_target_entity_type5"),
+ targetEntityId = Some("my_target_entity_id5"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = DateTime.now
+ )
+ val r6 = Event(
+ event = "my_event6",
+ entityType = "my_entity_type6",
+ entityId = "my_entity_id6",
+ targetEntityType = Some("my_target_entity_type6"),
+ targetEntityId = Some("my_target_entity_id6"),
+ properties = DataMap(
+ """{
+ "prop1" : 6,
+ "prop2" : "value2",
+ "prop3" : [6, 7, 8],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = DateTime.now
+ )
+
+ // timezone
+ val tz1 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id0",
+ targetEntityType = Some("my_target_entity_type"),
+ targetEntityId = Some("my_target_entity_id"),
+ properties = DataMap(
+ """{
+ "prop1" : 1,
+ "prop2" : "value2",
+ "prop3" : [1, 2, 3],
+ "prop4" : true,
+ "prop5" : ["a", "b", "c"],
+ "prop6" : 4.56
+ }"""
+ ),
+ eventTime = new DateTime(12345678, DateTimeZone.forID("-08:00")),
+ prId = Some("my_prid")
+ )
+
+ val tz2 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id1",
+ eventTime = new DateTime(12345678, DateTimeZone.forID("+02:00")),
+ prId = Some("my_prid")
+ )
+
+ val tz3 = Event(
+ event = "my_event",
+ entityType = "my_entity_type",
+ entityId = "my_entity_id2",
+ eventTime = new DateTime(12345678, DateTimeZone.forID("+08:00")),
+ prId = Some("my_prid")
+ )
+
+}
[4/7] incubator-predictionio git commit: Grand unification
Posted by do...@apache.org.
Grand unification
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/be8397fc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/be8397fc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/be8397fc
Branch: refs/heads/feature/xbuild
Commit: be8397fc63a967a5cabf838c52096bd806991db7
Parents: b6c90a0
Author: Donald Szeto <do...@apache.org>
Authored: Thu Mar 16 00:08:15 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Thu Mar 16 00:08:15 2017 -0700
----------------------------------------------------------------------
build.sbt | 177 +++++++++----------
common/build.sbt | 2 +
conf/pio-vendors.sh | 26 +++
conf/set_build_profile.sh | 31 ----
core/build.sbt | 2 +
data/build.sbt | 2 +
e2/build.sbt | 2 +
make-distribution.sh | 35 +---
project/Build.scala | 57 ------
project/PIOBuild.scala | 33 ++++
storage/elasticsearch/build.sbt | 2 +
storage/elasticsearch1/build.sbt | 2 +
storage/hbase/build.sbt | 2 +
storage/hdfs/build.sbt | 2 +
storage/jdbc/build.sbt | 2 +
storage/localfs/build.sbt | 2 +
tests/Dockerfile | 7 +-
tests/Dockerfile-es1 | 57 ------
tests/after_script.travis.sh | 8 +-
tests/build_docker.sh | 38 ++--
tests/docker-compose-es1.yml | 38 ----
tests/docker-compose.yml | 2 +-
tests/docker-files/env-conf/pio-env-es1.sh | 103 -----------
tests/docker-files/env-conf/pio-env.sh | 10 +-
.../engines/recommendation-engine/build.sbt | 4 +-
.../engines/recommendation-engine/manifest.json | 1 -
.../engines/recommendation-engine/pio.sbt | 4 -
.../recommendation-engine/project/pio-build.sbt | 1 -
tests/run_docker.sh | 19 +-
tests/script.travis.sh | 2 +-
tests/unit.sh | 13 +-
tools/build.sbt | 1 +
32 files changed, 222 insertions(+), 465 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/build.sbt
----------------------------------------------------------------------
diff --git a/build.sbt b/build.sbt
index 330dd06..16ce10e 100644
--- a/build.sbt
+++ b/build.sbt
@@ -15,55 +15,36 @@
* limitations under the License.
*/
+import PIOBuild._
import UnidocKeys._
-import scala.math.Ordering.Implicits._
-
-lazy val profiles: Map[String, Profile] =
- Map(
- "scala-2.10" -> Profile(
- name="scala-2.10",
- scalaVersion="2.10.5",
- sparkVersion="1.6.3",
- hadoopVersion="2.6.4",
- akkaVersion="2.3.15"),
-
- "scala-2.11" -> Profile(
- name="scala-2.11",
- scalaVersion="2.11.8",
- sparkVersion="2.0.2",
- hadoopVersion="2.7.3",
- akkaVersion="2.4.10"))
-
-lazy val defaultProfile = "scala-2.10"
-
-buildProfile := {
- val profileName = sys.props.get("build.profile").getOrElse(defaultProfile)
- val profile = profiles(profileName)
- val scalaVersion = profile.scalaVersion
-
- val sparkVersion = sys.props.get("spark.version") map { sv =>
- if ((versionMajor(sv), versionMinor(sv)) < (1, 6)) {
- throw new IllegalArgumentException("Spark versions below 1.6 are no longer supported")
- } else if (versionMajor(sv) >= 2 &&
- (versionMajor(scalaVersion), versionMinor(scalaVersion)) < (2, 11)) {
- throw new IllegalArgumentException("Spark 2.x requires Scala 2.11 and above")
- } else {
- sv
- }
- } getOrElse(profile.sparkVersion)
-
- val hadoopVersion = sys.props.get("hadoop.version").getOrElse(profile.hadoopVersion)
-
- if (hadoopVersion != profile.hadoopVersion || sparkVersion != profile.sparkVersion) {
- profile.copy(
- name = profile.name + "-custom",
- sparkVersion = sparkVersion,
- hadoopVersion = hadoopVersion)
- } else {
- profile
- }
-}
+lazy val scalaSparkDependenciesVersion = Map(
+ "2.10" -> Map(
+ "1.6" -> Map(
+ "akka" -> "2.3.15",
+ "hadoop" -> "2.6.5",
+ "json4s" -> "3.2.10"),
+ "2.0" -> Map(
+ "akka" -> "2.3.16",
+ "hadoop" -> "2.7.3",
+ "json4s" -> "3.2.11"),
+ "2.1" -> Map(
+ "akka" -> "2.3.16",
+ "hadoop" -> "2.7.3",
+ "json4s" -> "3.2.11")),
+ "2.11" -> Map(
+ "1.6" -> Map(
+ "akka" -> "2.3.15",
+ "hadoop" -> "2.6.5",
+ "json4s" -> "3.2.10"),
+ "2.0" -> Map(
+ "akka" -> "2.4.17",
+ "hadoop" -> "2.7.3",
+ "json4s" -> "3.2.11"),
+ "2.1" -> Map(
+ "akka" -> "2.4.17",
+ "hadoop" -> "2.7.3",
+ "json4s" -> "3.2.11")))
name := "apache-predictionio-parent"
@@ -71,13 +52,9 @@ version in ThisBuild := "0.11.0-SNAPSHOT"
organization in ThisBuild := "org.apache.predictionio"
-scalaVersion in ThisBuild := {
- val version = buildProfile.value.scalaVersion
- if (versionMinor(version) < 11) {
- sLog.value.warn(s"Scala version ${version} is deprecated!")
- }
- version
-}
+scalaVersion in ThisBuild := sys.props.getOrElse("scala.version", "2.10.6")
+
+crossScalaVersions in ThisBuild := Seq("2.10.6", "2.11.8")
scalacOptions in ThisBuild ++= Seq("-deprecation", "-unchecked", "-feature")
@@ -87,13 +64,22 @@ fork in (ThisBuild, run) := true
javacOptions in (ThisBuild, compile) ++= Seq("-source", "1.7", "-target", "1.7",
"-Xlint:deprecation", "-Xlint:unchecked")
-akkaVersion in ThisBuild := buildProfile.value.akkaVersion
+// Ignore differentiation of Spark patch levels
+sparkVersion in ThisBuild := sys.props.getOrElse("spark.version", "1.6.3")
+
+akkaVersion in ThisBuild := sys.props.getOrElse(
+ "akka.version",
+ scalaSparkDependenciesVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("akka"))
+
+lazy val es = sys.props.getOrElse("elasticsearch.version", "1.7.6")
-json4sVersion in ThisBuild := "3.2.10"
+elasticsearchVersion in ThisBuild := es
-sparkVersion in ThisBuild := buildProfile.value.sparkVersion
+json4sVersion in ThisBuild := scalaSparkDependenciesVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("json4s")
-hadoopVersion in ThisBuild := buildProfile.value.hadoopVersion
+hadoopVersion in ThisBuild := sys.props.getOrElse(
+ "hadoop.version",
+ scalaSparkDependenciesVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("hadoop"))
val pioBuildInfoSettings = buildInfoSettings ++ Seq(
sourceGenerators in Compile <+= buildInfo,
@@ -115,7 +101,7 @@ val genjavadocSettings: Seq[sbt.Def.Setting[_]] = Seq(
// in unit tests of data subproject are used later in spark jobs executed in core.
// They need to have properly configured classloaders to load core classes for spark
// in subsequent tests.
-def coreClasses(baseDirectory: java.io.File, scalaVersion: String) = Seq(
+def coreClasses(baseDirectory: java.io.File, scalaVersion: String): Seq[File] = Seq(
baseDirectory / s"../core/target/scala-${versionPrefix(scalaVersion)}/classes",
baseDirectory / s"../core/target/scala-${versionPrefix(scalaVersion)}/test-classes")
@@ -127,45 +113,54 @@ val commonSettings = Seq(
val common = (project in file("common")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ disablePlugins(sbtassembly.AssemblyPlugin)
val data = (project in file("data")).
dependsOn(common).
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
settings(unmanagedSourceDirectories in Compile +=
- sourceDirectory.value / s"main/spark-${versionMajor(sparkVersion.value)}")
+ sourceDirectory.value / s"main/spark-${versionMajor(sparkVersion.value)}").
+ disablePlugins(sbtassembly.AssemblyPlugin)
val dataElasticsearch1 = (project in file("storage/elasticsearch1")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ settings(publishArtifact := false)
val dataElasticsearch = (project in file("storage/elasticsearch")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ settings(publishArtifact := false)
val dataHbase = (project in file("storage/hbase")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ settings(publishArtifact := false)
val dataHdfs = (project in file("storage/hdfs")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ settings(publishArtifact := false)
val dataJdbc = (project in file("storage/jdbc")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ settings(publishArtifact := false)
val dataLocalfs = (project in file("storage/localfs")).
settings(commonSettings: _*).
- settings(genjavadocSettings: _*)
+ settings(genjavadocSettings: _*).
+ settings(publishArtifact := false)
val core = (project in file("core")).
dependsOn(data).
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
settings(pioBuildInfoSettings: _*).
- enablePlugins(SbtTwirl)
+ enablePlugins(SbtTwirl).
+ disablePlugins(sbtassembly.AssemblyPlugin)
val tools = (project in file("tools")).
dependsOn(core).
@@ -173,12 +168,28 @@ val tools = (project in file("tools")).
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
enablePlugins(SbtTwirl).
- settings(fullClasspath in Test ++= coreClasses(baseDirectory.value, scalaVersion.value))
+ settings(fullClasspath in Test ++= coreClasses(baseDirectory.value, scalaVersion.value)).
+ settings(publishArtifact := false)
val e2 = (project in file("e2")).
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
- settings(fullClasspath in Test ++= coreClasses(baseDirectory.value, scalaVersion.value))
+ settings(fullClasspath in Test ++= coreClasses(baseDirectory.value, scalaVersion.value)).
+ disablePlugins(sbtassembly.AssemblyPlugin)
+
+val dataEs = if (versionMajor(es) == 1) dataElasticsearch1 else dataElasticsearch
+
+val subprojects = Seq(
+ common,
+ core,
+ data,
+ dataEs,
+ dataHbase,
+ dataHdfs,
+ dataJdbc,
+ dataLocalfs,
+ tools,
+ e2)
val root = (project in file(".")).
settings(commonSettings: _*).
@@ -237,18 +248,8 @@ val root = (project in file(".")).
"docs/javadoc/javadoc-overview.html",
"-noqualifier",
"java.lang")).
- aggregate(
- common,
- core,
- data,
- dataElasticsearch1,
- dataElasticsearch,
- dataHbase,
- dataHdfs,
- dataJdbc,
- dataLocalfs,
- tools,
- e2)
+ aggregate(subprojects map Project.projectToRef: _*).
+ disablePlugins(sbtassembly.AssemblyPlugin)
val pioUnidoc = taskKey[Unit]("Builds PredictionIO ScalaDoc")
@@ -308,11 +309,9 @@ parallelExecution in Global := false
testOptions in Test += Tests.Argument("-oDF")
-printProfile := {
- val profile = buildProfile.value
- println(s"PIO_PROFILE_VERSION=${profile.name}")
- println(s"PIO_SCALA_VERSION=${profile.scalaVersion}")
- println(s"PIO_SPARK_VERSION=${profile.sparkVersion}")
- println(s"PIO_HADOOP_VERSION=${profile.hadoopVersion}")
- println(s"PIO_AKKA_VERSION=${profile.akkaVersion}")
+printBuildInfo := {
+ println(s"PIO_SCALA_VERSION=${scalaVersion.value}")
+ println(s"PIO_SPARK_VERSION=${sparkVersion.value}")
+ println(s"PIO_ELASTICSEARCH_VERSION=${elasticsearchVersion.value}")
+ println(s"PIO_HADOOP_VERSION=${hadoopVersion.value}")
}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/common/build.sbt
----------------------------------------------------------------------
diff --git a/common/build.sbt b/common/build.sbt
index 47fba3b..5a0fba1 100644
--- a/common/build.sbt
+++ b/common/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-common"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/conf/pio-vendors.sh
----------------------------------------------------------------------
diff --git a/conf/pio-vendors.sh b/conf/pio-vendors.sh
index 00e7783..830b576 100644
--- a/conf/pio-vendors.sh
+++ b/conf/pio-vendors.sh
@@ -19,6 +19,32 @@
# IMPORTANT: PIO_*_VERSION for dependencies must be set before envoking this script.
# `source conf/set_build_profile.sh $BUILD_PROFILE` to get the proper versions
+if [ -z "$PIO_SCALA_VERSION" ]; then
+ PIO_SCALA_VERSION="2.10.6"
+fi
+
+if [ -z "$PIO_SPARK_VERSION" ]; then
+ PIO_SPARK_VERSION="1.6.3"
+fi
+
+if [ -z "$PIO_HADOOP_VERSION" ]; then
+ PIO_HADOOP_VERSION="2.6.5"
+fi
+
+if [ -z "$PIO_ELASTICSEARCH_VERSION" ]; then
+ PIO_ELASTICSEARCH_VERSION="1.7.3"
+fi
+
+ES_MAJOR=`echo $PIO_ELASTICSEARCH_VERSION | awk -F. '{print $1}'`
+
+if [ "$ES_MAJOR" = "1" ]; then
+ export ES_IMAGE="elasticsearch"
+ export ES_TAG="1"
+else
+ export ES_IMAGE="docker.elastic.co/elasticsearch/elasticsearch"
+ export ES_TAG="5.2.2"
+fi
+
PGSQL_JAR=postgresql-9.4-1204.jdbc41.jar
PGSQL_DOWNLOAD=https://jdbc.postgresql.org/download/${PGSQL_JAR}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/conf/set_build_profile.sh
----------------------------------------------------------------------
diff --git a/conf/set_build_profile.sh b/conf/set_build_profile.sh
deleted file mode 100755
index 397e824..0000000
--- a/conf/set_build_profile.sh
+++ /dev/null
@@ -1,31 +0,0 @@
-#!/bin/bash -
-#
-# 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.
-#
-
-# Sets version of profile dependencies from sbt configuration.
-# eg. Run `source ./set_build_profile.sh scala-2.11`
-
-set -e
-
-if [[ "$#" -ne 1 ]]; then
- echo "Usage: set-build-profile.sh <build-profile>"
- exit 1
-fi
-
-set -a
-eval `sbt/sbt --error 'set showSuccess := false' -Dbuild.profile=$1 printProfile | grep '.*_VERSION=.*'`
-set +a
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/core/build.sbt
----------------------------------------------------------------------
diff --git a/core/build.sbt b/core/build.sbt
index 584133c..2d2a381 100644
--- a/core/build.sbt
+++ b/core/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-core"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/data/build.sbt
----------------------------------------------------------------------
diff --git a/data/build.sbt b/data/build.sbt
index 27eeb78..778a8f1 100644
--- a/data/build.sbt
+++ b/data/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/e2/build.sbt
----------------------------------------------------------------------
diff --git a/e2/build.sbt b/e2/build.sbt
index 50de5e5..f30536a 100644
--- a/e2/build.sbt
+++ b/e2/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-e2"
parallelExecution in Test := false
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/make-distribution.sh
----------------------------------------------------------------------
diff --git a/make-distribution.sh b/make-distribution.sh
index 2000340..70ba4a4 100755
--- a/make-distribution.sh
+++ b/make-distribution.sh
@@ -21,15 +21,12 @@ set -e
usage ()
{
- echo "Usage: $0 [-h|--help] [--with-es=x]"
+ echo "Usage: $0 [-h|--help]"
echo ""
echo " -h|--help Show usage"
- echo ""
- echo " --with-es=1 Build distribution with Elasticsearch 1 support as default"
- echo " --with-es=5 Build distribution with Elasticsearch 5 support as default"
}
-ES_VERSION=1
+JAVA_PROPS=()
for i in "$@"
do
@@ -39,11 +36,8 @@ case $i in
shift
exit
;;
- --with-es=*)
- ES_VERSION="${i#*=}"
- shift
- ;;
-D*)
+ JAVA_PROPS+=("$i")
shift
;;
*)
@@ -53,14 +47,6 @@ case $i in
esac
done
-if [ "$ES_VERSION" = "1" ] || [ "$ES_VERSION" = "5" ]
-then
- echo -e "\033[0;32mBuilding with Elasticsearch $ES_VERSION support as the default choice\033[0m"
-else
- usage
- exit 1
-fi
-
FWDIR="$(cd `dirname $0`; pwd)"
DISTDIR="${FWDIR}/dist"
@@ -68,10 +54,10 @@ VERSION=$(grep ^version ${FWDIR}/build.sbt | grep ThisBuild | grep -o '".*"' | s
echo "Building binary distribution for PredictionIO $VERSION..."
+sbt/sbt printBuildInfo
+
cd ${FWDIR}
-sbt/sbt "$@" common/publishLocal data/publishLocal core/publishLocal e2/publishLocal \
-dataElasticsearch1/assembly dataElasticsearch/assembly dataHbase/assembly dataHdfs/assembly \
-dataJdbc/assembly dataLocalfs/assembly tools/assembly
+sbt/sbt "$JAVA_PROPS" publishLocal assembly
cd ${FWDIR}
rm -rf ${DISTDIR}
@@ -79,8 +65,8 @@ mkdir -p ${DISTDIR}/bin
mkdir -p ${DISTDIR}/conf
mkdir -p ${DISTDIR}/lib
mkdir -p ${DISTDIR}/lib/spark
-mkdir -p ${DISTDIR}/lib/extra
mkdir -p ${DISTDIR}/project
+
mkdir -p ${DISTDIR}/sbt
cp ${FWDIR}/bin/* ${DISTDIR}/bin || :
@@ -90,13 +76,6 @@ cp ${FWDIR}/sbt/sbt ${DISTDIR}/sbt
cp ${FWDIR}/assembly/*assembly*jar ${DISTDIR}/lib
cp ${FWDIR}/assembly/spark/*jar ${DISTDIR}/lib/spark
-if [ "$ES_VERSION" = "5" ]
-then
- mv ${DISTDIR}/lib/spark/pio-data-elasticsearch1-assembly-*.jar ${DISTDIR}/lib/extra
-else
- mv ${DISTDIR}/lib/spark/pio-data-elasticsearch-assembly-*.jar ${DISTDIR}/lib/extra
-fi
-
rm -f ${DISTDIR}/lib/*javadoc.jar
rm -f ${DISTDIR}/lib/*sources.jar
rm -f ${DISTDIR}/conf/pio-env.sh
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/project/Build.scala
----------------------------------------------------------------------
diff --git a/project/Build.scala b/project/Build.scala
deleted file mode 100644
index cc37912..0000000
--- a/project/Build.scala
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-import sbt._
-import Keys._
-
-case class Profile(
- name: String,
- scalaVersion: String,
- sparkVersion: String,
- hadoopVersion: String,
- akkaVersion: String)
-
-object PIOBuild extends Build {
- val elasticsearchVersion = SettingKey[String](
- "elasticsearch-version",
- "The version of Elasticsearch used for building.")
- val json4sVersion = SettingKey[String](
- "json4s-version",
- "The version of JSON4S used for building.")
- val sparkVersion = SettingKey[String](
- "spark-version",
- "The version of Apache Spark used for building.")
- val hadoopVersion = SettingKey[String](
- "hadoop-version",
- "The version of Apache Hadoop used for building")
- val akkaVersion = SettingKey[String](
- "akka-version",
- "The version of Akka used for building")
- val buildProfile = SettingKey[Profile](
- "build-profile",
- "The dependency profile used for the build")
- val childrenPomExtra = SettingKey[scala.xml.NodeSeq](
- "children-pom-extra",
- "Extra POM data for children projects.")
-
- def versionPrefix(versionString: String) =
- versionString.split('.').take(2).mkString(".")
- def versionMajor(versionString: String) = versionString.split('.')(0).toInt
- def versionMinor(versionString: String) = versionString.split('.')(1).toInt
-
- lazy val printProfile = taskKey[Unit]("Print settings for the chosen profile")
-}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/project/PIOBuild.scala
----------------------------------------------------------------------
diff --git a/project/PIOBuild.scala b/project/PIOBuild.scala
new file mode 100644
index 0000000..69ac189
--- /dev/null
+++ b/project/PIOBuild.scala
@@ -0,0 +1,33 @@
+/*
+ * 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.
+ */
+
+import sbt._
+
+object PIOBuild {
+ val elasticsearchVersion = settingKey[String]("The version of Elasticsearch used for building")
+ val json4sVersion = settingKey[String]("The version of JSON4S used for building")
+ val sparkVersion = settingKey[String]("The version of Apache Spark used for building")
+ val hadoopVersion = settingKey[String]("The version of Apache Hadoop used for building")
+ val akkaVersion = settingKey[String]("The version of Akka used for building")
+ val childrenPomExtra = settingKey[scala.xml.NodeSeq]("Extra POM data for children projects")
+
+ def versionPrefix(versionString: String): String = versionString.split('.').take(2).mkString(".")
+ def versionMajor(versionString: String): Int = versionString.split('.')(0).toInt
+ def versionMinor(versionString: String): Int = versionString.split('.')(1).toInt
+
+ lazy val printBuildInfo = taskKey[Unit]("Print build information")
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/storage/elasticsearch/build.sbt
----------------------------------------------------------------------
diff --git a/storage/elasticsearch/build.sbt b/storage/elasticsearch/build.sbt
index 7d9b2ad..a7d3ab5 100644
--- a/storage/elasticsearch/build.sbt
+++ b/storage/elasticsearch/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data-elasticsearch"
elasticsearchVersion := "5.2.1"
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/storage/elasticsearch1/build.sbt
----------------------------------------------------------------------
diff --git a/storage/elasticsearch1/build.sbt b/storage/elasticsearch1/build.sbt
index dde7285..c78c75d 100644
--- a/storage/elasticsearch1/build.sbt
+++ b/storage/elasticsearch1/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data-elasticsearch1"
elasticsearchVersion := "1.7.3"
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/storage/hbase/build.sbt
----------------------------------------------------------------------
diff --git a/storage/hbase/build.sbt b/storage/hbase/build.sbt
index 513e294..22b53d8 100644
--- a/storage/hbase/build.sbt
+++ b/storage/hbase/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data-hbase"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/storage/hdfs/build.sbt
----------------------------------------------------------------------
diff --git a/storage/hdfs/build.sbt b/storage/hdfs/build.sbt
index 7ddc86c..2942e56 100644
--- a/storage/hdfs/build.sbt
+++ b/storage/hdfs/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data-hdfs"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/storage/jdbc/build.sbt
----------------------------------------------------------------------
diff --git a/storage/jdbc/build.sbt b/storage/jdbc/build.sbt
index 2011722..905ed53 100644
--- a/storage/jdbc/build.sbt
+++ b/storage/jdbc/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data-jdbc"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/storage/localfs/build.sbt
----------------------------------------------------------------------
diff --git a/storage/localfs/build.sbt b/storage/localfs/build.sbt
index f9e101d..90a61db 100644
--- a/storage/localfs/build.sbt
+++ b/storage/localfs/build.sbt
@@ -15,6 +15,8 @@
* limitations under the License.
*/
+import PIOBuild._
+
name := "apache-predictionio-data-localfs"
libraryDependencies ++= Seq(
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/Dockerfile
----------------------------------------------------------------------
diff --git a/tests/Dockerfile b/tests/Dockerfile
index a04b79c..619bdf8 100644
--- a/tests/Dockerfile
+++ b/tests/Dockerfile
@@ -20,13 +20,15 @@ FROM predictionio/pio
ARG SPARK_ARCHIVE
ARG SPARK_DIR
ARG PGSQL_JAR
-ARG BUILD_PROFILE
ARG PIO_SCALA_VERSION
ARG PIO_SPARK_VERSION
+ARG PIO_ELASTICSEARCH_VERSION
-ENV BUILD_PROFILE=$BUILD_PROFILE
ENV PIO_SCALA_VERSION=$PIO_SCALA_VERSION
ENV PIO_SPARK_VERSION=$PIO_SPARK_VERSION
+ENV PIO_ELASTICSEARCH_VERSION=$PIO_ELASTICSEARCH_VERSION
+
+ENV PGSQL_JAR=$PGSQL_JAR
# WORKAROUND: es-hadoop stops on RDD#take(1)
ADD docker-files/${SPARK_ARCHIVE} /vendors
@@ -35,7 +37,6 @@ ENV SPARK_HOME /vendors/${SPARK_DIR}
COPY docker-files/${PGSQL_JAR} /drivers/${PGSQL_JAR}
COPY docker-files/init.sh init.sh
-COPY docker-files/set_build_profile.sh set_build_profile.sh
COPY docker-files/env-conf/hbase-site.xml ${PIO_HOME}/conf/hbase-site.xml
COPY docker-files/env-conf/pio-env.sh ${PIO_HOME}/conf/pio-env.sh
COPY docker-files/pgpass /root/.pgpass
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/Dockerfile-es1
----------------------------------------------------------------------
diff --git a/tests/Dockerfile-es1 b/tests/Dockerfile-es1
deleted file mode 100644
index 73d4c36..0000000
--- a/tests/Dockerfile-es1
+++ /dev/null
@@ -1,57 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-FROM predictionio/pio
-
-ENV SPARK_VERSION 1.6.3
-ENV ELASTICSEARCH_VERSION 1.7
-ENV HBASE_VERSION 1.0.0
-
-ADD docker-files/spark-${SPARK_VERSION}-bin-hadoop2.6.tgz /vendors
-# WORKAROUND: es-hadoop stops on RDD#take(1)
-RUN echo "spark.locality.wait.node 0s" > /vendors/spark-${SPARK_VERSION}-bin-hadoop2.6/conf/spark-defaults.conf
-ENV SPARK_HOME /vendors/spark-${SPARK_VERSION}-bin-hadoop2.6
-
-COPY docker-files/postgresql-9.4-1204.jdbc41.jar /drivers/postgresql-9.4-1204.jdbc41.jar
-COPY docker-files/init.sh init.sh
-COPY docker-files/env-conf/hbase-site.xml ${PIO_HOME}/conf/hbase-site.xml
-COPY docker-files/env-conf/pio-env-es1.sh ${PIO_HOME}/conf/pio-env.sh
-COPY docker-files/pgpass /root/.pgpass
-RUN chmod 600 /root/.pgpass
-
-# Python
-RUN pip install python-dateutil
-RUN pip install pytz
-
-# Default repositories setup
-ENV PIO_STORAGE_REPOSITORIES_METADATA_SOURCE PGSQL
-ENV PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE PGSQL
-ENV PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE PGSQL
-
-# JVM settings
-ENV JVM_OPTS '-Dfile.encoding=UTF8 -Xms2048M -Xmx2048M -Xss8M -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=256M'
-
-# Expose relevant ports
-# pio engine
-EXPOSE 8000
-# eventserver
-EXPOSE 7070
-
-ENV SLEEP_TIME 30
-
-ENTRYPOINT ["/init.sh"]
-CMD 'bash'
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/after_script.travis.sh
----------------------------------------------------------------------
diff --git a/tests/after_script.travis.sh b/tests/after_script.travis.sh
index 8cbe2c0..e9d5792 100755
--- a/tests/after_script.travis.sh
+++ b/tests/after_script.travis.sh
@@ -18,12 +18,10 @@
DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
+source $DIR/../conf/pio-vendors.sh
+
# Print a summary of containers used
docker ps -a
# Clean up used containers
-if [ "$ES_VERSION" = "1" ]; then
- docker-compose -f $DIR/docker-compose-es1.yml down
-else
- docker-compose -f $DIR/docker-compose.yml down
-fi
+docker-compose -f $DIR/docker-compose.yml down
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/build_docker.sh
----------------------------------------------------------------------
diff --git a/tests/build_docker.sh b/tests/build_docker.sh
index 13a6f6a..9b6c2c6 100755
--- a/tests/build_docker.sh
+++ b/tests/build_docker.sh
@@ -22,7 +22,6 @@ docker pull predictionio/pio-testing-base
pushd $DIR/..
-source conf/set_build_profile.sh ${BUILD_PROFILE}
source conf/pio-vendors.sh
if [ ! -f $DIR/docker-files/${PGSQL_JAR} ]; then
wget $PGSQL_DOWNLOAD
@@ -33,36 +32,23 @@ if [ ! -f $DIR/docker-files/${SPARK_ARCHIVE} ]; then
mv $SPARK_ARCHIVE $DIR/docker-files/
fi
-if [ -z "$ES_VERSION" ]; then
- ./make-distribution.sh -Dbuild.profile=${BUILD_PROFILE}
-else
- ./make-distribution.sh --with-es=$ES_VERSION -Dbuild.profile=${BUILD_PROFILE}
-fi
+./make-distribution.sh \
+ -Dscala.version=$PIO_SCALA_VERSION \
+ -Dspark.version=$PIO_SPARK_VERSION \
+ -Dhadoop.version=$PIO_HADOOP_VERSION \
+ -Delasticsearch.version=$PIO_ELASTICSEARCH_VERSION
sbt/sbt clean
mkdir assembly
cp dist/lib/*.jar assembly/
mkdir -p lib/spark
-if [ -e dist/lib/spark/*.jar ]; then
- cp dist/lib/spark/*.jar lib/spark
-fi
+cp dist/lib/spark/*.jar lib/spark
rm *.tar.gz
docker build -t predictionio/pio .
popd
-if [ "$ES_VERSION" = "1" ]; then
- docker build -t predictionio/pio-testing-es1 -f $DIR/Dockerfile-es1 $DIR \
- --build-arg SPARK_ARCHIVE=$SPARK_ARCHIVE \
- --build-arg SPARK_DIR=$SPARK_DIR \
- --build-arg PGSQL_JAR=$PGSQL_JAR \
- --build-arg BUILD_PROFILE=$BUILD_PROFILE \
- --build-arg PIO_SCALA_VERSION=$PIO_SCALA_VERSION \
- --build-arg PIO_SPARK_VERSION=$PIO_SPARK_VERSION
-else
- docker build -t predictionio/pio-testing $DIR \
- --build-arg SPARK_ARCHIVE=$SPARK_ARCHIVE \
- --build-arg SPARK_DIR=$SPARK_DIR \
- --build-arg PGSQL_JAR=$PGSQL_JAR \
- --build-arg BUILD_PROFILE=$BUILD_PROFILE \
- --build-arg PIO_SCALA_VERSION=$PIO_SCALA_VERSION \
- --build-arg PIO_SPARK_VERSION=$PIO_SPARK_VERSION
-fi
+docker build -t predictionio/pio-testing $DIR \
+ --build-arg SPARK_ARCHIVE=$SPARK_ARCHIVE \
+ --build-arg SPARK_DIR=$SPARK_DIR \
+ --build-arg PGSQL_JAR=$PGSQL_JAR \
+ --build-arg PIO_SCALA_VERSION=$PIO_SCALA_VERSION \
+ --build-arg PIO_SPARK_VERSION=$PIO_SPARK_VERSION
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/docker-compose-es1.yml
----------------------------------------------------------------------
diff --git a/tests/docker-compose-es1.yml b/tests/docker-compose-es1.yml
deleted file mode 100644
index 381f41c..0000000
--- a/tests/docker-compose-es1.yml
+++ /dev/null
@@ -1,38 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-
-version: "2"
-services:
- elasticsearch:
- image: elasticsearch:1
- environment:
- - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
- hbase:
- image: harisekhon/hbase:1.0
- postgres:
- image: postgres:9
- environment:
- POSTGRES_USER: pio
- POSTGRES_PASSWORD: pio
- POSTGRES_INITDB_ARGS: --encoding=UTF8
- pio-testing:
- image: predictionio/pio-testing-es1:latest
- depends_on:
- - elasticsearch
- - hbase
- - postgres
- volumes:
- - ~/.ivy2:/root/.ivy2
- - ~/.sbt:/root/.sbt
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/docker-compose.yml
----------------------------------------------------------------------
diff --git a/tests/docker-compose.yml b/tests/docker-compose.yml
index 06126a5..ac1d91d 100644
--- a/tests/docker-compose.yml
+++ b/tests/docker-compose.yml
@@ -16,7 +16,7 @@
version: "2"
services:
elasticsearch:
- image: docker.elastic.co/elasticsearch/elasticsearch:5.2.2
+ image: ${ES_IMAGE}:${ES_TAG}
environment:
- xpack.security.enabled=false
- "ES_JAVA_OPTS=-Xms512m -Xmx512m"
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/docker-files/env-conf/pio-env-es1.sh
----------------------------------------------------------------------
diff --git a/tests/docker-files/env-conf/pio-env-es1.sh b/tests/docker-files/env-conf/pio-env-es1.sh
deleted file mode 100644
index e1076ba..0000000
--- a/tests/docker-files/env-conf/pio-env-es1.sh
+++ /dev/null
@@ -1,103 +0,0 @@
-#!/usr/bin/env bash
-#
-# Copy this file as pio-env.sh and edit it for your site's configuration.
-#
-# 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.
-#
-
-# PredictionIO Main Configuration
-#
-# This section controls core behavior of PredictionIO. It is very likely that
-# you need to change these to fit your site.
-
-# SPARK_HOME: Apache Spark is a hard dependency and must be configured.
-SPARK_HOME=$SPARK_HOME
-
-POSTGRES_JDBC_DRIVER=/drivers/postgresql-9.4-1204.jdbc41.jar
-MYSQL_JDBC_DRIVER=
-
-# ES_CONF_DIR: You must configure this if you have advanced configuration for
-# your Elasticsearch setup.
-# ES_CONF_DIR=/opt/elasticsearch
-
-# HADOOP_CONF_DIR: You must configure this if you intend to run PredictionIO
-# with Hadoop 2.
-# HADOOP_CONF_DIR=/opt/hadoop
-
-# HBASE_CONF_DIR: You must configure this if you intend to run PredictionIO
-# with HBase on a remote cluster.
-HBASE_CONF_DIR=$PIO_HOME/conf
-
-# Filesystem paths where PredictionIO uses as block storage.
-PIO_FS_BASEDIR=$HOME/.pio_store
-PIO_FS_ENGINESDIR=$PIO_FS_BASEDIR/engines
-PIO_FS_TMPDIR=$PIO_FS_BASEDIR/tmp
-
-# PredictionIO Storage Configuration
-#
-# This section controls programs that make use of PredictionIO's built-in
-# storage facilities. Default values are shown below.
-#
-# For more information on storage configuration please refer to
-# https://docs.prediction.io/system/anotherdatastore/
-
-# Storage Repositories
-
-# Default is to use PostgreSQL
-PIO_STORAGE_REPOSITORIES_METADATA_NAME=pio_meta
-PIO_STORAGE_REPOSITORIES_METADATA_SOURCE=$PIO_STORAGE_REPOSITORIES_METADATA_SOURCE
-
-PIO_STORAGE_REPOSITORIES_EVENTDATA_NAME=pio_event
-PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE=$PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE
-
-PIO_STORAGE_REPOSITORIES_MODELDATA_NAME=pio_model
-PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE=$PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE
-
-# Storage Data Sources
-
-# PostgreSQL Default Settings
-# Please change "pio" to your database name in PIO_STORAGE_SOURCES_PGSQL_URL
-# Please change PIO_STORAGE_SOURCES_PGSQL_USERNAME and
-# PIO_STORAGE_SOURCES_PGSQL_PASSWORD accordingly
-PIO_STORAGE_SOURCES_PGSQL_TYPE=jdbc
-PIO_STORAGE_SOURCES_PGSQL_URL=jdbc:postgresql://postgres/pio
-PIO_STORAGE_SOURCES_PGSQL_USERNAME=pio
-PIO_STORAGE_SOURCES_PGSQL_PASSWORD=pio
-
-# MySQL Example
-# PIO_STORAGE_SOURCES_MYSQL_TYPE=jdbc
-# PIO_STORAGE_SOURCES_MYSQL_URL=jdbc:mysql://localhost/pio
-# PIO_STORAGE_SOURCES_MYSQL_USERNAME=pio
-# PIO_STORAGE_SOURCES_MYSQL_PASSWORD=pio
-
-# Elasticsearch Example
-PIO_STORAGE_SOURCES_ELASTICSEARCH_TYPE=elasticsearch
-#PIO_STORAGE_SOURCES_ELASTICSEARCH_CLUSTERNAME=pio
-PIO_STORAGE_SOURCES_ELASTICSEARCH_HOSTS=elasticsearch
-PIO_STORAGE_SOURCES_ELASTICSEARCH_PORTS=9300
-#PIO_STORAGE_SOURCES_ELASTICSEARCH_HOME=$ELASTICSEARCH_HOME
-
-# Local File System Example
-PIO_STORAGE_SOURCES_LOCALFS_TYPE=localfs
-PIO_STORAGE_SOURCES_LOCALFS_PATH=$PIO_FS_BASEDIR/local_models
-
-# HBase Example
-PIO_STORAGE_SOURCES_HBASE_TYPE=hbase
-#PIO_STORAGE_SOURCES_HBASE_HOME=$HBASE_HOME
-
-# HDFS config
-PIO_STORAGE_SOURCES_HDFS_TYPE=hdfs
-PIO_STORAGE_SOURCES_HDFS_PATH=/hdfs_models
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/docker-files/env-conf/pio-env.sh
----------------------------------------------------------------------
diff --git a/tests/docker-files/env-conf/pio-env.sh b/tests/docker-files/env-conf/pio-env.sh
index 0acf3a7..51d2e05 100644
--- a/tests/docker-files/env-conf/pio-env.sh
+++ b/tests/docker-files/env-conf/pio-env.sh
@@ -24,9 +24,9 @@
# you need to change these to fit your site.
# SPARK_HOME: Apache Spark is a hard dependency and must be configured.
-SPARK_HOME=$SPARK_HOME
+# SPARK_HOME=$SPARK_HOME
-POSTGRES_JDBC_DRIVER=/drivers/postgresql-9.4-1204.jdbc41.jar
+POSTGRES_JDBC_DRIVER=/drivers/$PGSQL_JAR
MYSQL_JDBC_DRIVER=
# ES_CONF_DIR: You must configure this if you have advanced configuration for
@@ -88,7 +88,11 @@ PIO_STORAGE_SOURCES_ELASTICSEARCH_TYPE=elasticsearch
#PIO_STORAGE_SOURCES_ELASTICSEARCH_CLUSTERNAME=pio
PIO_STORAGE_SOURCES_ELASTICSEARCH_HOSTS=elasticsearch
PIO_STORAGE_SOURCES_ELASTICSEARCH_SCHEMES=http
-PIO_STORAGE_SOURCES_ELASTICSEARCH_PORTS=9200
+if [ "$PIO_ELASTICSEARCH_VERSION" == "1" ]; then
+ PIO_STORAGE_SOURCES_ELASTICSEARCH_PORTS=9300
+else
+ PIO_STORAGE_SOURCES_ELASTICSEARCH_PORTS=9200
+fi
#PIO_STORAGE_SOURCES_ELASTICSEARCH_HOME=$ELASTICSEARCH_HOME
# Local File System Example
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/pio_tests/engines/recommendation-engine/build.sbt
----------------------------------------------------------------------
diff --git a/tests/pio_tests/engines/recommendation-engine/build.sbt b/tests/pio_tests/engines/recommendation-engine/build.sbt
index dcb9fb6..13fe354 100644
--- a/tests/pio_tests/engines/recommendation-engine/build.sbt
+++ b/tests/pio_tests/engines/recommendation-engine/build.sbt
@@ -27,5 +27,5 @@ organization := "org.apache.predictionio"
libraryDependencies ++= Seq(
"org.apache.predictionio" %% "apache-predictionio-core" % "0.11.0-SNAPSHOT" % "provided",
- "org.apache.spark" %% "spark-core" % sys.env.getOrElse("PIO_SPARK_VERSION", "1.6.2") % "provided",
- "org.apache.spark" %% "spark-mllib" % sys.env.getOrElse("PIO_SPARK_VERSION", "1.6.2") % "provided")
+ "org.apache.spark" %% "spark-core" % sys.env.getOrElse("PIO_SPARK_VERSION", "1.6.3") % "provided",
+ "org.apache.spark" %% "spark-mllib" % sys.env.getOrElse("PIO_SPARK_VERSION", "1.6.3") % "provided")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/pio_tests/engines/recommendation-engine/manifest.json
----------------------------------------------------------------------
diff --git a/tests/pio_tests/engines/recommendation-engine/manifest.json b/tests/pio_tests/engines/recommendation-engine/manifest.json
deleted file mode 100644
index 9c545ce..0000000
--- a/tests/pio_tests/engines/recommendation-engine/manifest.json
+++ /dev/null
@@ -1 +0,0 @@
-{"id":"yDBpzjz39AjIxlOAh8W4t3QSc75uPCuJ","version":"ee98ff9c009ef0d9fb1bc6b78750b83a0ceb37b2","name":"recommendation-engine","description":"pio-autogen-manifest","files":[],"engineFactory":""}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/pio_tests/engines/recommendation-engine/pio.sbt
----------------------------------------------------------------------
diff --git a/tests/pio_tests/engines/recommendation-engine/pio.sbt b/tests/pio_tests/engines/recommendation-engine/pio.sbt
deleted file mode 100644
index 5d8c366..0000000
--- a/tests/pio_tests/engines/recommendation-engine/pio.sbt
+++ /dev/null
@@ -1,4 +0,0 @@
-// Generated automatically by pio build.
-// Changes in this file will be overridden.
-
-pioVersion := "0.11.0-SNAPSHOT"
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/pio_tests/engines/recommendation-engine/project/pio-build.sbt
----------------------------------------------------------------------
diff --git a/tests/pio_tests/engines/recommendation-engine/project/pio-build.sbt b/tests/pio_tests/engines/recommendation-engine/project/pio-build.sbt
deleted file mode 100644
index 8346a96..0000000
--- a/tests/pio_tests/engines/recommendation-engine/project/pio-build.sbt
+++ /dev/null
@@ -1 +0,0 @@
-addSbtPlugin("io.prediction" % "pio-build" % "0.9.0")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/run_docker.sh
----------------------------------------------------------------------
diff --git a/tests/run_docker.sh b/tests/run_docker.sh
index 6257fa2..ad7e189 100755
--- a/tests/run_docker.sh
+++ b/tests/run_docker.sh
@@ -46,17 +46,10 @@ shift
DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
-if [ "$ES_VERSION" = "1" ]; then
- docker-compose -f $DIR/docker-compose-es1.yml run \
- -e PIO_STORAGE_REPOSITORIES_METADATA_SOURCE=$META \
- -e PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE=$EVENT \
- -e PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE=$MODEL \
- pio-testing $@
-else
- docker-compose -f $DIR/docker-compose.yml run \
- -e PIO_STORAGE_REPOSITORIES_METADATA_SOURCE=$META \
- -e PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE=$EVENT \
- -e PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE=$MODEL \
- pio-testing $@
-fi
+source $DIR/../conf/pio-vendors.sh
+docker-compose -f $DIR/docker-compose.yml run \
+ -e PIO_STORAGE_REPOSITORIES_METADATA_SOURCE=$META \
+ -e PIO_STORAGE_REPOSITORIES_EVENTDATA_SOURCE=$EVENT \
+ -e PIO_STORAGE_REPOSITORIES_MODELDATA_SOURCE=$MODEL \
+ pio-testing $@
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/script.travis.sh
----------------------------------------------------------------------
diff --git a/tests/script.travis.sh b/tests/script.travis.sh
index efdbb9e..1d4985e 100755
--- a/tests/script.travis.sh
+++ b/tests/script.travis.sh
@@ -18,7 +18,7 @@
if [[ $BUILD_TYPE == Unit ]]; then
tests/run_docker.sh $METADATA_REP $EVENTDATA_REP $MODELDATA_REP \
- "/PredictionIO/tests/unit.sh $BUILD_PROFILE"
+ "/PredictionIO/tests/unit.sh"
else
tests/run_docker.sh $METADATA_REP $EVENTDATA_REP $MODELDATA_REP \
"python3 /PredictionIO/tests/pio_tests/tests.py"
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tests/unit.sh
----------------------------------------------------------------------
diff --git a/tests/unit.sh b/tests/unit.sh
index 724b878..d6572c2 100755
--- a/tests/unit.sh
+++ b/tests/unit.sh
@@ -25,11 +25,20 @@ pushd /PredictionIO
set -a
source ./conf/pio-env.sh
set +a
+source ./conf/pio-vendors.sh
# Run stylecheck
-sbt/sbt -Dbuild.profile=$1 scalastyle
+sbt/sbt scalastyle \
+ -Dscala.version=$PIO_SCALA_VERSION \
+ -Dspark.version=$PIO_SPARK_VERSION \
+ -Dhadoop.version=$PIO_HADOOP_VERSION \
+ -Delasticsearch.version=$PIO_ELASTICSEARCH_VERSION
# Run all unit tests
-sbt/sbt -Dbuild.profile=$1 test
+sbt/sbt test \
+ -Dscala.version=$PIO_SCALA_VERSION \
+ -Dspark.version=$PIO_SPARK_VERSION \
+ -Dhadoop.version=$PIO_HADOOP_VERSION \
+ -Delasticsearch.version=$PIO_ELASTICSEARCH_VERSION
popd
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/be8397fc/tools/build.sbt
----------------------------------------------------------------------
diff --git a/tools/build.sbt b/tools/build.sbt
index d1ce2b9..0971b62 100644
--- a/tools/build.sbt
+++ b/tools/build.sbt
@@ -15,6 +15,7 @@
* limitations under the License.
*/
+import PIOBuild._
import sbtassembly.AssemblyPlugin.autoImport._
name := "apache-predictionio-tools"
[7/7] incubator-predictionio git commit: Fix SBT tests
Posted by do...@apache.org.
Fix SBT tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/d051216a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/d051216a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/d051216a
Branch: refs/heads/feature/xbuild
Commit: d051216a7cea133ba24c470d7e2a03f07a1ae5cf
Parents: f928333
Author: Donald Szeto <do...@apache.org>
Authored: Thu Mar 16 07:51:53 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Thu Mar 16 07:51:53 2017 -0700
----------------------------------------------------------------------
build.sbt | 47 +++++++++++++++++++--------------------------
core/build.sbt | 3 ++-
storage/jdbc/build.sbt | 2 +-
tools/build.sbt | 1 +
4 files changed, 24 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/d051216a/build.sbt
----------------------------------------------------------------------
diff --git a/build.sbt b/build.sbt
index 16ce10e..38b5f9e 100644
--- a/build.sbt
+++ b/build.sbt
@@ -18,7 +18,7 @@
import PIOBuild._
import UnidocKeys._
-lazy val scalaSparkDependenciesVersion = Map(
+lazy val scalaSparkDepsVersion = Map(
"2.10" -> Map(
"1.6" -> Map(
"akka" -> "2.3.15",
@@ -69,17 +69,17 @@ sparkVersion in ThisBuild := sys.props.getOrElse("spark.version", "1.6.3")
akkaVersion in ThisBuild := sys.props.getOrElse(
"akka.version",
- scalaSparkDependenciesVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("akka"))
+ scalaSparkDepsVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("akka"))
lazy val es = sys.props.getOrElse("elasticsearch.version", "1.7.6")
elasticsearchVersion in ThisBuild := es
-json4sVersion in ThisBuild := scalaSparkDependenciesVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("json4s")
+json4sVersion in ThisBuild := scalaSparkDepsVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("json4s")
hadoopVersion in ThisBuild := sys.props.getOrElse(
"hadoop.version",
- scalaSparkDependenciesVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("hadoop"))
+ scalaSparkDepsVersion(versionPrefix(scalaVersion.value))(versionPrefix(sparkVersion.value))("hadoop"))
val pioBuildInfoSettings = buildInfoSettings ++ Seq(
sourceGenerators in Compile <+= buildInfo,
@@ -97,33 +97,12 @@ val genjavadocSettings: Seq[sbt.Def.Setting[_]] = Seq(
libraryDependencies += compilerPlugin("com.typesafe.genjavadoc" %% "genjavadoc-plugin" % "0.10" cross CrossVersion.full),
scalacOptions <+= target map (t => "-P:genjavadoc:out=" + (t / "java")))
-// Paths specified below are required for the tests, since thread pools initialized
-// in unit tests of data subproject are used later in spark jobs executed in core.
-// They need to have properly configured classloaders to load core classes for spark
-// in subsequent tests.
-def coreClasses(baseDirectory: java.io.File, scalaVersion: String): Seq[File] = Seq(
- baseDirectory / s"../core/target/scala-${versionPrefix(scalaVersion)}/classes",
- baseDirectory / s"../core/target/scala-${versionPrefix(scalaVersion)}/test-classes")
-
val conf = file("conf")
val commonSettings = Seq(
autoAPIMappings := true,
unmanagedClasspath in Test += conf)
-val common = (project in file("common")).
- settings(commonSettings: _*).
- settings(genjavadocSettings: _*).
- disablePlugins(sbtassembly.AssemblyPlugin)
-
-val data = (project in file("data")).
- dependsOn(common).
- settings(commonSettings: _*).
- settings(genjavadocSettings: _*).
- settings(unmanagedSourceDirectories in Compile +=
- sourceDirectory.value / s"main/spark-${versionMajor(sparkVersion.value)}").
- disablePlugins(sbtassembly.AssemblyPlugin)
-
val dataElasticsearch1 = (project in file("storage/elasticsearch1")).
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
@@ -154,8 +133,23 @@ val dataLocalfs = (project in file("storage/localfs")).
settings(genjavadocSettings: _*).
settings(publishArtifact := false)
+val common = (project in file("common")).
+ settings(commonSettings: _*).
+ settings(genjavadocSettings: _*).
+ disablePlugins(sbtassembly.AssemblyPlugin)
+
+val data = (project in file("data")).
+ dependsOn(common).
+ dependsOn(dataJdbc % "test->compile").
+ settings(commonSettings: _*).
+ settings(genjavadocSettings: _*).
+ settings(unmanagedSourceDirectories in Compile +=
+ sourceDirectory.value / s"main/spark-${versionMajor(sparkVersion.value)}").
+ disablePlugins(sbtassembly.AssemblyPlugin)
+
val core = (project in file("core")).
dependsOn(data).
+ dependsOn(dataJdbc % "test->compile").
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
settings(pioBuildInfoSettings: _*).
@@ -165,16 +159,15 @@ val core = (project in file("core")).
val tools = (project in file("tools")).
dependsOn(core).
dependsOn(data).
+ dependsOn(dataJdbc % "test->compile").
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
enablePlugins(SbtTwirl).
- settings(fullClasspath in Test ++= coreClasses(baseDirectory.value, scalaVersion.value)).
settings(publishArtifact := false)
val e2 = (project in file("e2")).
settings(commonSettings: _*).
settings(genjavadocSettings: _*).
- settings(fullClasspath in Test ++= coreClasses(baseDirectory.value, scalaVersion.value)).
disablePlugins(sbtassembly.AssemblyPlugin)
val dataEs = if (versionMajor(es) == 1) dataElasticsearch1 else dataElasticsearch
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/d051216a/core/build.sbt
----------------------------------------------------------------------
diff --git a/core/build.sbt b/core/build.sbt
index 2d2a381..c6bac3f 100644
--- a/core/build.sbt
+++ b/core/build.sbt
@@ -37,8 +37,9 @@ libraryDependencies ++= Seq(
"org.json4s" %% "json4s-native" % json4sVersion.value,
"org.json4s" %% "json4s-ext" % json4sVersion.value,
"org.scalaj" %% "scalaj-http" % "1.1.6",
- "org.scalatest" %% "scalatest" % "2.1.7" % "test",
"org.slf4j" % "slf4j-log4j12" % "1.7.18",
+ "org.postgresql" % "postgresql" % "9.4-1204-jdbc41" % "test",
+ "org.scalatest" %% "scalatest" % "2.1.7" % "test",
"org.specs2" %% "specs2" % "2.3.13" % "test")
parallelExecution in Test := false
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/d051216a/storage/jdbc/build.sbt
----------------------------------------------------------------------
diff --git a/storage/jdbc/build.sbt b/storage/jdbc/build.sbt
index 905ed53..b0ff2fd 100644
--- a/storage/jdbc/build.sbt
+++ b/storage/jdbc/build.sbt
@@ -23,8 +23,8 @@ libraryDependencies ++= Seq(
"org.apache.predictionio" %% "apache-predictionio-core" % version.value % "provided",
"org.apache.predictionio" %% "apache-predictionio-data" % version.value % "provided",
"org.apache.spark" %% "spark-sql" % sparkVersion.value % "provided",
- "org.postgresql" % "postgresql" % "9.4-1204-jdbc41",
"org.scalikejdbc" %% "scalikejdbc" % "2.3.5",
+ "org.postgresql" % "postgresql" % "9.4-1204-jdbc41" % "test",
"org.scalatest" %% "scalatest" % "2.1.7" % "test",
"org.specs2" %% "specs2" % "2.3.13" % "test")
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/d051216a/tools/build.sbt
----------------------------------------------------------------------
diff --git a/tools/build.sbt b/tools/build.sbt
index 0971b62..b61a20a 100644
--- a/tools/build.sbt
+++ b/tools/build.sbt
@@ -36,6 +36,7 @@ libraryDependencies ++= Seq(
"com.typesafe.akka" %% "akka-actor" % akkaVersion.value,
"com.typesafe.akka" %% "akka-slf4j" % akkaVersion.value,
"io.spray" %% "spray-testkit" % "1.3.3" % "test",
+ "org.postgresql" % "postgresql" % "9.4-1204-jdbc41" % "test",
"org.specs2" %% "specs2" % "2.3.13" % "test")
dependencyOverrides += "org.slf4j" % "slf4j-log4j12" % "1.7.18"
[6/7] incubator-predictionio git commit: Put back to the right place
Posted by do...@apache.org.
Put back to the right place
Project: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/commit/f9283330
Tree: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/tree/f9283330
Diff: http://git-wip-us.apache.org/repos/asf/incubator-predictionio/diff/f9283330
Branch: refs/heads/feature/xbuild
Commit: f92833304a7bfc467a9143c9274ac4ee8022462c
Parents: ab11980
Author: Donald Szeto <do...@apache.org>
Authored: Thu Mar 16 00:15:39 2017 -0700
Committer: Donald Szeto <do...@apache.org>
Committed: Thu Mar 16 00:15:39 2017 -0700
----------------------------------------------------------------------
.../predictionio/data/view/PBatchView.scala | 204 ++++++++++++++++++
.../predictionio/data/view/PBatchView.scala | 212 -------------------
2 files changed, 204 insertions(+), 212 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/f9283330/data/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala
----------------------------------------------------------------------
diff --git a/data/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala b/data/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala
new file mode 100644
index 0000000..cbdebce
--- /dev/null
+++ b/data/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package org.apache.predictionio.data.view
+
+import org.apache.predictionio.data.storage.{DataMap, Event, EventValidation, Storage}
+import org.apache.spark.SparkContext
+import org.apache.spark.rdd.RDD
+import org.joda.time.DateTime
+import org.json4s.JValue
+
+
+// each JValue data associated with the time it is set
+private[predictionio] case class PropTime(val d: JValue, val t: Long) extends Serializable
+
+private[predictionio] case class SetProp (
+ val fields: Map[String, PropTime],
+ // last set time. Note: fields could be empty with valid set time
+ val t: Long) extends Serializable {
+
+ def ++ (that: SetProp): SetProp = {
+ val commonKeys = fields.keySet.intersect(that.fields.keySet)
+
+ val common: Map[String, PropTime] = commonKeys.map { k =>
+ val thisData = this.fields(k)
+ val thatData = that.fields(k)
+ // only keep the value with latest time
+ val v = if (thisData.t > thatData.t) thisData else thatData
+ (k, v)
+ }.toMap
+
+ val combinedFields = common ++
+ (this.fields -- commonKeys) ++ (that.fields -- commonKeys)
+
+ // keep the latest set time
+ val combinedT = if (this.t > that.t) this.t else that.t
+
+ SetProp(
+ fields = combinedFields,
+ t = combinedT
+ )
+ }
+}
+
+private[predictionio] case class UnsetProp (fields: Map[String, Long]) extends Serializable {
+ def ++ (that: UnsetProp): UnsetProp = {
+ val commonKeys = fields.keySet.intersect(that.fields.keySet)
+
+ val common: Map[String, Long] = commonKeys.map { k =>
+ val thisData = this.fields(k)
+ val thatData = that.fields(k)
+ // only keep the value with latest time
+ val v = if (thisData > thatData) thisData else thatData
+ (k, v)
+ }.toMap
+
+ val combinedFields = common ++
+ (this.fields -- commonKeys) ++ (that.fields -- commonKeys)
+
+ UnsetProp(
+ fields = combinedFields
+ )
+ }
+}
+
+private[predictionio] case class DeleteEntity (t: Long) extends Serializable {
+ def ++ (that: DeleteEntity): DeleteEntity = {
+ if (this.t > that.t) this else that
+ }
+}
+
+private[predictionio] case class EventOp (
+ val setProp: Option[SetProp] = None,
+ val unsetProp: Option[UnsetProp] = None,
+ val deleteEntity: Option[DeleteEntity] = None
+) extends Serializable {
+
+ def ++ (that: EventOp): EventOp = {
+ EventOp(
+ setProp = (setProp ++ that.setProp).reduceOption(_ ++ _),
+ unsetProp = (unsetProp ++ that.unsetProp).reduceOption(_ ++ _),
+ deleteEntity = (deleteEntity ++ that.deleteEntity).reduceOption(_ ++ _)
+ )
+ }
+
+ def toDataMap(): Option[DataMap] = {
+ setProp.flatMap { set =>
+
+ val unsetKeys: Set[String] = unsetProp.map( unset =>
+ unset.fields.filter{ case (k, v) => (v >= set.fields(k).t) }.keySet
+ ).getOrElse(Set())
+
+ val combinedFields = deleteEntity.map { delete =>
+ if (delete.t >= set.t) {
+ None
+ } else {
+ val deleteKeys: Set[String] = set.fields
+ .filter { case (k, PropTime(kv, t)) =>
+ (delete.t >= t)
+ }.keySet
+ Some(set.fields -- unsetKeys -- deleteKeys)
+ }
+ }.getOrElse{
+ Some(set.fields -- unsetKeys)
+ }
+
+ // Note: mapValues() doesn't return concrete Map and causes
+ // NotSerializableException issue. Use map(identity) to work around this.
+ // see https://issues.scala-lang.org/browse/SI-7005
+ combinedFields.map(f => DataMap(f.mapValues(_.d).map(identity)))
+ }
+ }
+
+}
+
+private[predictionio] object EventOp {
+ def apply(e: Event): EventOp = {
+ val t = e.eventTime.getMillis
+ e.event match {
+ case "$set" => {
+ val fields = e.properties.fields.mapValues(jv =>
+ PropTime(jv, t)
+ ).map(identity)
+
+ EventOp(
+ setProp = Some(SetProp(fields = fields, t = t))
+ )
+ }
+ case "$unset" => {
+ val fields = e.properties.fields.mapValues(jv => t).map(identity)
+ EventOp(
+ unsetProp = Some(UnsetProp(fields = fields))
+ )
+ }
+ case "$delete" => {
+ EventOp(
+ deleteEntity = Some(DeleteEntity(t))
+ )
+ }
+ case _ => {
+ EventOp()
+ }
+ }
+ }
+}
+
+@deprecated("Use PEvents or PEventStore instead.", "0.9.2")
+class PBatchView(
+ val appId: Int,
+ val startTime: Option[DateTime],
+ val untilTime: Option[DateTime],
+ val sc: SparkContext) {
+
+ // NOTE: parallel Events DB interface
+ @transient lazy val eventsDb = Storage.getPEvents()
+
+ @transient lazy val _events: RDD[Event] =
+ eventsDb.getByAppIdAndTimeAndEntity(
+ appId = appId,
+ startTime = startTime,
+ untilTime = untilTime,
+ entityType = None,
+ entityId = None)(sc)
+
+ // TODO: change to use EventSeq?
+ @transient lazy val events: RDD[Event] = _events
+
+ def aggregateProperties(
+ entityType: String,
+ startTimeOpt: Option[DateTime] = None,
+ untilTimeOpt: Option[DateTime] = None
+ ): RDD[(String, DataMap)] = {
+
+ _events
+ .filter( e => ((e.entityType == entityType) &&
+ (EventValidation.isSpecialEvents(e.event))) )
+ .map( e => (e.entityId, EventOp(e) ))
+ .aggregateByKey[EventOp](EventOp())(
+ // within same partition
+ seqOp = { case (u, v) => u ++ v },
+ // across partition
+ combOp = { case (accu, u) => accu ++ u }
+ )
+ .mapValues(_.toDataMap)
+ .filter{ case (k, v) => v.isDefined }
+ .map{ case (k, v) => (k, v.get) }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-predictionio/blob/f9283330/storage/hbase/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala
----------------------------------------------------------------------
diff --git a/storage/hbase/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala b/storage/hbase/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala
deleted file mode 100644
index b453820..0000000
--- a/storage/hbase/src/main/scala/org/apache/predictionio/data/view/PBatchView.scala
+++ /dev/null
@@ -1,212 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-
-package org.apache.predictionio.data.view
-
-import org.apache.predictionio.data.storage.hbase.HBPEvents
-import org.apache.predictionio.data.storage.Event
-import org.apache.predictionio.data.storage.EventValidation
-import org.apache.predictionio.data.storage.DataMap
-import org.apache.predictionio.data.storage.Storage
-
-import org.joda.time.DateTime
-
-import org.json4s.JValue
-
-import org.apache.spark.SparkContext
-import org.apache.spark.SparkContext._
-import org.apache.spark.rdd.RDD
-
-
-// each JValue data associated with the time it is set
-private[predictionio] case class PropTime(val d: JValue, val t: Long) extends Serializable
-
-private[predictionio] case class SetProp (
- val fields: Map[String, PropTime],
- // last set time. Note: fields could be empty with valid set time
- val t: Long) extends Serializable {
-
- def ++ (that: SetProp): SetProp = {
- val commonKeys = fields.keySet.intersect(that.fields.keySet)
-
- val common: Map[String, PropTime] = commonKeys.map { k =>
- val thisData = this.fields(k)
- val thatData = that.fields(k)
- // only keep the value with latest time
- val v = if (thisData.t > thatData.t) thisData else thatData
- (k, v)
- }.toMap
-
- val combinedFields = common ++
- (this.fields -- commonKeys) ++ (that.fields -- commonKeys)
-
- // keep the latest set time
- val combinedT = if (this.t > that.t) this.t else that.t
-
- SetProp(
- fields = combinedFields,
- t = combinedT
- )
- }
-}
-
-private[predictionio] case class UnsetProp (fields: Map[String, Long]) extends Serializable {
- def ++ (that: UnsetProp): UnsetProp = {
- val commonKeys = fields.keySet.intersect(that.fields.keySet)
-
- val common: Map[String, Long] = commonKeys.map { k =>
- val thisData = this.fields(k)
- val thatData = that.fields(k)
- // only keep the value with latest time
- val v = if (thisData > thatData) thisData else thatData
- (k, v)
- }.toMap
-
- val combinedFields = common ++
- (this.fields -- commonKeys) ++ (that.fields -- commonKeys)
-
- UnsetProp(
- fields = combinedFields
- )
- }
-}
-
-private[predictionio] case class DeleteEntity (t: Long) extends Serializable {
- def ++ (that: DeleteEntity): DeleteEntity = {
- if (this.t > that.t) this else that
- }
-}
-
-private[predictionio] case class EventOp (
- val setProp: Option[SetProp] = None,
- val unsetProp: Option[UnsetProp] = None,
- val deleteEntity: Option[DeleteEntity] = None
-) extends Serializable {
-
- def ++ (that: EventOp): EventOp = {
- EventOp(
- setProp = (setProp ++ that.setProp).reduceOption(_ ++ _),
- unsetProp = (unsetProp ++ that.unsetProp).reduceOption(_ ++ _),
- deleteEntity = (deleteEntity ++ that.deleteEntity).reduceOption(_ ++ _)
- )
- }
-
- def toDataMap(): Option[DataMap] = {
- setProp.flatMap { set =>
-
- val unsetKeys: Set[String] = unsetProp.map( unset =>
- unset.fields.filter{ case (k, v) => (v >= set.fields(k).t) }.keySet
- ).getOrElse(Set())
-
- val combinedFields = deleteEntity.map { delete =>
- if (delete.t >= set.t) {
- None
- } else {
- val deleteKeys: Set[String] = set.fields
- .filter { case (k, PropTime(kv, t)) =>
- (delete.t >= t)
- }.keySet
- Some(set.fields -- unsetKeys -- deleteKeys)
- }
- }.getOrElse{
- Some(set.fields -- unsetKeys)
- }
-
- // Note: mapValues() doesn't return concrete Map and causes
- // NotSerializableException issue. Use map(identity) to work around this.
- // see https://issues.scala-lang.org/browse/SI-7005
- combinedFields.map(f => DataMap(f.mapValues(_.d).map(identity)))
- }
- }
-
-}
-
-private[predictionio] object EventOp {
- def apply(e: Event): EventOp = {
- val t = e.eventTime.getMillis
- e.event match {
- case "$set" => {
- val fields = e.properties.fields.mapValues(jv =>
- PropTime(jv, t)
- ).map(identity)
-
- EventOp(
- setProp = Some(SetProp(fields = fields, t = t))
- )
- }
- case "$unset" => {
- val fields = e.properties.fields.mapValues(jv => t).map(identity)
- EventOp(
- unsetProp = Some(UnsetProp(fields = fields))
- )
- }
- case "$delete" => {
- EventOp(
- deleteEntity = Some(DeleteEntity(t))
- )
- }
- case _ => {
- EventOp()
- }
- }
- }
-}
-
-@deprecated("Use PEvents or PEventStore instead.", "0.9.2")
-class PBatchView(
- val appId: Int,
- val startTime: Option[DateTime],
- val untilTime: Option[DateTime],
- val sc: SparkContext) {
-
- // NOTE: parallel Events DB interface
- @transient lazy val eventsDb = Storage.getPEvents()
-
- @transient lazy val _events: RDD[Event] =
- eventsDb.getByAppIdAndTimeAndEntity(
- appId = appId,
- startTime = startTime,
- untilTime = untilTime,
- entityType = None,
- entityId = None)(sc)
-
- // TODO: change to use EventSeq?
- @transient lazy val events: RDD[Event] = _events
-
- def aggregateProperties(
- entityType: String,
- startTimeOpt: Option[DateTime] = None,
- untilTimeOpt: Option[DateTime] = None
- ): RDD[(String, DataMap)] = {
-
- _events
- .filter( e => ((e.entityType == entityType) &&
- (EventValidation.isSpecialEvents(e.event))) )
- .map( e => (e.entityId, EventOp(e) ))
- .aggregateByKey[EventOp](EventOp())(
- // within same partition
- seqOp = { case (u, v) => u ++ v },
- // across partition
- combOp = { case (accu, u) => accu ++ u }
- )
- .mapValues(_.toDataMap)
- .filter{ case (k, v) => v.isDefined }
- .map{ case (k, v) => (k, v.get) }
- }
-
-}