mvn 测试出错：java.lang.IllegalStateException：无法在已停止的 SparkContext 上调用方法
Error with mvn test: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext
仅当 运行 使用 mvn test 连接所有测试用例时，我才收到以下错误。如果我 运行 每个测试 类 独立于 IDE，则不会发生这种情况。我正在使用 https://github.com/MrPowers/spark-fast-tests 框架。
特质
trait SparkSessionTestWrapper {
lazy val spark: SparkSession = {
SparkSession
.builder()
.master("local")
.appName("spark-fast-tests test session")
.config("spark.sql.shuffle.partitions", "1")
.getOrCreate()
}
}
测试
WordCountDSAppTestSpec
class WordCountDSAppTestSpec extends FlatSpec with SparkSessionTestWrapper with DatasetComparer {
import spark.implicits._
"toWords" should "split the file into words" in {
val sourceDf = Seq(
("one"),
("two"),
(""),
("three Three")
).toDF("line").as[Line]
val expectedDF = Seq(
("one", "one"),
("two", "two"),
("three Three", "three"),
("three Three", "Three"),
("", "")
).toDF("line", "word").as[LineAndWord]
val actualDF = WordCountDSApp.toWords(sourceDf)
assertSmallDatasetEquality(actualDF, expectedDF, orderedComparison = false)
}
"countWords" should "return count of each word" in {
val wordsDF = Seq(
("one", "one"),
("two", "two"),
("three Three", "three"),
("three Three", "Three"),
("", "")
).toDF("line", "word").as[LineAndWord]
val tupleEncoder = Encoders.tuple(Encoders.STRING, Encoders.LONG)
val expectedDF = Seq(
("one", 1L),
("two", 1L),
("three", 2L)
).toDF("value", "count(1)").as[(String, Long)]
val actualDF = WordCountDSApp.countWords(wordsDF)
assertSmallDatasetEquality(actualDF, expectedDF, orderedComparison = false)
}
}
ProductSalesAppTestSpec
class ProductSalesAppTestSpec extends FlatSpec with SparkSessionTestWrapper with DatasetComparer {
import spark.implicits._
val productCols = Seq("product_id", "product_name", "price")
val productDF = Seq(
("0", "product_0", "22"),
("1", "product_1", "30"),
("2", "product_2", "91")
).toDF(productCols: _*)
productDF.createOrReplaceTempView("PRODUCTS")
val orderCols = Seq("order_id", "product_id", "seller_id", "date", "num_pieces_sold", "bill_raw_text")
"howManyProductsHaveBeenSoldAtLeastOnce" should "return zero when no orders are made for products" in {
val orderDF = Seq(
("1", "10", "0", "2020-07-10", "26", "kyeibuumwlyhuwksx"),
("2", "20", "0", "2020-07-08", "13", "kyeibuumwlyhuwksx"),
("3", "30", "0", "2020-07-05", "38", "kyeibuumwlyhuwksx"),
("4", "40", "0", "2020-07-05", "56", "kyeibuumwlyhuwksx")
).toDF(orderCols: _*)
orderDF.createOrReplaceTempView("ORDERS")
assert(ProductSalesApp.howManyProductsHaveBeenAoldAtLeastOnce(spark) == 0)
}
}
错误
- should return zero when no orders are made for products *** FAILED ***
org.apache.spark.SparkException: Exception thrown in awaitResult:
at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:226)
at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.doExecuteBroadcast(BroadcastExchangeExec.scala:146)
at org.apache.spark.sql.execution.InputAdapter.doExecuteBroadcast(WholeStageCodegenExec.scala:387)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeBroadcast.apply(SparkPlan.scala:144)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeBroadcast.apply(SparkPlan.scala:140)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.executeBroadcast(SparkPlan.scala:140)
at org.apache.spark.sql.execution.joins.BroadcastHashJoinExec.prepareBroadcast(BroadcastHashJoinExec.scala:117)
...
Cause: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
This stopped SparkContext was created at:
org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:926)
com.aravind.oss.SparkSessionTestWrapper$class.spark(SparkSessionTestWrapper.scala:16)
com.aravind.oss.eg.spark.sales.ProductSalesAppTestSpec.spark$lzycompute(ProductSalesAppTestSpec.scala:13)
com.aravind.oss.eg.spark.sales.ProductSalesAppTestSpec.spark(ProductSalesAppTestSpec.scala:13)
com.aravind.oss.eg.spark.sales.ProductSalesAppTestSpec.<init>(ProductSalesAppTestSpec.scala:18)
sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
java.lang.reflect.Constructor.newInstance(Constructor.java:422)
java.lang.Class.newInstance(Class.java:442)
org.scalatest.tools.DiscoverySuite$.getSuiteInstance(DiscoverySuite.scala:66)
org.scalatest.tools.DiscoverySuite$$anonfun.apply(DiscoverySuite.scala:38)
org.scalatest.tools.DiscoverySuite$$anonfun.apply(DiscoverySuite.scala:37)
scala.collection.TraversableLike$$anonfun$map.apply(TraversableLike.scala:234)
scala.collection.TraversableLike$$anonfun$map.apply(TraversableLike.scala:234)
scala.collection.Iterator$class.foreach(Iterator.scala:891)
scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
scala.collection.AbstractIterable.foreach(Iterable.scala:54)
scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
The currently active SparkContext was created at:
org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:926)
com.aravind.oss.SparkSessionTestWrapper$class.spark(SparkSessionTestWrapper.scala:16)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec.spark$lzycompute(WordCountDSAppTestSpec.scala:15)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec.spark(WordCountDSAppTestSpec.scala:15)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec$$anonfun.apply$mcV$sp(WordCountDSAppTestSpec.scala:20)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec$$anonfun.apply(WordCountDSAppTestSpec.scala:19)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec$$anonfun.apply(WordCountDSAppTestSpec.scala:19)
org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
org.scalatest.Transformer.apply(Transformer.scala:22)
org.scalatest.Transformer.apply(Transformer.scala:20)
org.scalatest.FlatSpecLike$$anon.apply(FlatSpecLike.scala:1682)
org.scalatest.TestSuite$class.withFixture(TestSuite.scala:196)
org.scalatest.FlatSpec.withFixture(FlatSpec.scala:1685)
org.scalatest.FlatSpecLike$class.invokeWithFixture(FlatSpecLike.scala:1679)
org.scalatest.FlatSpecLike$$anonfun$runTest.apply(FlatSpecLike.scala:1692)
org.scalatest.FlatSpecLike$$anonfun$runTest.apply(FlatSpecLike.scala:1692)
org.scalatest.SuperEngine.runTestImpl(Engine.scala:286)
org.scalatest.FlatSpecLike$class.runTest(FlatSpecLike.scala:1692)
org.scalatest.FlatSpec.runTest(FlatSpec.scala:1685)
at org.apache.spark.SparkContext.assertNotStopped(SparkContext.scala:100)
at org.apache.spark.SparkContext.defaultParallelism(SparkContext.scala:2359)
at org.apache.spark.sql.execution.LocalTableScanExec.numParallelism$lzycompute(LocalTableScanExec.scala:49)
at org.apache.spark.sql.execution.LocalTableScanExec.numParallelism(LocalTableScanExec.scala:48)
at org.apache.spark.sql.execution.LocalTableScanExec.rdd$lzycompute(LocalTableScanExec.scala:51)
at org.apache.spark.sql.execution.LocalTableScanExec.rdd(LocalTableScanExec.scala:51)
at org.apache.spark.sql.execution.LocalTableScanExec.doExecute(LocalTableScanExec.scala:55)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery.apply(SparkPlan.scala:155)
...
我检查了你的git repo
进入这个测试用例后
"howManyProductsHaveBeenSoldAtLeastOnce" should "return zero when no orders are made for products" in {
当您使用来自 trait 的 spark 会话时，它正在通过 mvn test 使用新线程重置（停止 spark 上下文）您的 spark 会话。
来自 Itellij IDE 它在同一个线程中并且它不重置 spark（不停止上下文）
我能够在 mvn 调试模式 (mvn -DforkMode=never test -f pom.xml) 和使用 spark.sparkContext.isStopped 下进行调查，如下面的代码所示。
这就是它从 intellij 而不是通过 mvn test
工作的原因
所以我放弃了为 spark session 创建 trait 的想法，并创建了一个本地 spark session（也许你可以尝试 import org.scalatest.{BeforeAndAfterAll}）以获得更好的代码。
package com.aravind.oss.eg.spark.wordcount
import com.aravind.oss.SparkSessionTestWrapper
import com.aravind.oss.eg.spark.sales.ProductSalesApp
import com.github.mrpowers.spark.fast.tests.DatasetComparer
import org.scalatest.{BeforeAndAfterAll, FlatSpec}
import ProductSalesApp._
import org.apache.spark.sql.SparkSession
class ProductSalesAppTestSpec extends FlatSpec with DatasetComparer {
lazy implicit val spark: SparkSession =
SparkSession
.builder()
.master("local")
.appName("spark-fast-tests test session")
.config("spark.sql.shuffle.partitions", "1")
.getOrCreate()
import spark.implicits._
val productCols = Seq("product_id", "product_name", "price")
val productDF = Seq(
("0", "product_0", "22"),
("1", "product_1", "30"),
("2", "product_2", "91")
).toDF(productCols: _*)
productDF.createOrReplaceTempView("PRODUCTS")
val orderCols = Seq("order_id", "product_id", "seller_id", "date", "num_pieces_sold", "bill_raw_text")
val orderDF = Seq(
("1", "10", "0", "2020-07-10", "26", "kyeibuumwlyhuwksx"),
("2", "20", "0", "2020-07-08", "13", "kyeibuumwlyhuwksx"),
("3", "30", "0", "2020-07-05", "38", "kyeibuumwlyhuwksx"),
("4", "40", "0", "2020-07-05", "56", "kyeibuumwlyhuwksx")
).toDF(orderCols: _*)
orderDF.createOrReplaceTempView("ORDERS")
orderDF.show
println( spark.sparkContext.isStopped)
println( spark.sparkContext.isLocal)
"howManyProductsHaveBeenSoldAtLeastOnce" should "return zero when no orders are made for products" in {
println( "howManyProductsHaveBeenSoldAtLeastOnce "+ spark.sparkContext.isStopped)
println( spark.sparkContext.isLocal)
println(ProductSalesApp.howManyProductsHaveBeenAoldAtLeastOnce(spark))
assert(ProductSalesApp.howManyProductsHaveBeenAoldAtLeastOnce(spark) == 0)
}
}
结果：
+--------+----------+---------+----------+---------------+-----------------+
|order_id|product_id|seller_id| date|num_pieces_sold| bill_raw_text|
+--------+----------+---------+----------+---------------+-----------------+
| 1| 10| 0|2020-07-10| 26|kyeibuumwlyhuwksx|
| 2| 20| 0|2020-07-08| 13|kyeibuumwlyhuwksx|
| 3| 30| 0|2020-07-05| 38|kyeibuumwlyhuwksx|
| 4| 40| 0|2020-07-05| 56|kyeibuumwlyhuwksx|
+--------+----------+---------+----------+---------------+-----------------+
false
true
howManyProductsHaveBeenSoldAtLeastOnce false
true
0
仅当 运行 使用 mvn test 连接所有测试用例时，我才收到以下错误。如果我 运行 每个测试 类 独立于 IDE，则不会发生这种情况。我正在使用 https://github.com/MrPowers/spark-fast-tests 框架。
特质
trait SparkSessionTestWrapper {
lazy val spark: SparkSession = {
SparkSession
.builder()
.master("local")
.appName("spark-fast-tests test session")
.config("spark.sql.shuffle.partitions", "1")
.getOrCreate()
}
}
测试
WordCountDSAppTestSpec
class WordCountDSAppTestSpec extends FlatSpec with SparkSessionTestWrapper with DatasetComparer {
import spark.implicits._
"toWords" should "split the file into words" in {
val sourceDf = Seq(
("one"),
("two"),
(""),
("three Three")
).toDF("line").as[Line]
val expectedDF = Seq(
("one", "one"),
("two", "two"),
("three Three", "three"),
("three Three", "Three"),
("", "")
).toDF("line", "word").as[LineAndWord]
val actualDF = WordCountDSApp.toWords(sourceDf)
assertSmallDatasetEquality(actualDF, expectedDF, orderedComparison = false)
}
"countWords" should "return count of each word" in {
val wordsDF = Seq(
("one", "one"),
("two", "two"),
("three Three", "three"),
("three Three", "Three"),
("", "")
).toDF("line", "word").as[LineAndWord]
val tupleEncoder = Encoders.tuple(Encoders.STRING, Encoders.LONG)
val expectedDF = Seq(
("one", 1L),
("two", 1L),
("three", 2L)
).toDF("value", "count(1)").as[(String, Long)]
val actualDF = WordCountDSApp.countWords(wordsDF)
assertSmallDatasetEquality(actualDF, expectedDF, orderedComparison = false)
}
}
ProductSalesAppTestSpec
class ProductSalesAppTestSpec extends FlatSpec with SparkSessionTestWrapper with DatasetComparer {
import spark.implicits._
val productCols = Seq("product_id", "product_name", "price")
val productDF = Seq(
("0", "product_0", "22"),
("1", "product_1", "30"),
("2", "product_2", "91")
).toDF(productCols: _*)
productDF.createOrReplaceTempView("PRODUCTS")
val orderCols = Seq("order_id", "product_id", "seller_id", "date", "num_pieces_sold", "bill_raw_text")
"howManyProductsHaveBeenSoldAtLeastOnce" should "return zero when no orders are made for products" in {
val orderDF = Seq(
("1", "10", "0", "2020-07-10", "26", "kyeibuumwlyhuwksx"),
("2", "20", "0", "2020-07-08", "13", "kyeibuumwlyhuwksx"),
("3", "30", "0", "2020-07-05", "38", "kyeibuumwlyhuwksx"),
("4", "40", "0", "2020-07-05", "56", "kyeibuumwlyhuwksx")
).toDF(orderCols: _*)
orderDF.createOrReplaceTempView("ORDERS")
assert(ProductSalesApp.howManyProductsHaveBeenAoldAtLeastOnce(spark) == 0)
}
}
错误
- should return zero when no orders are made for products *** FAILED ***
org.apache.spark.SparkException: Exception thrown in awaitResult:
at org.apache.spark.util.ThreadUtils$.awaitResult(ThreadUtils.scala:226)
at org.apache.spark.sql.execution.exchange.BroadcastExchangeExec.doExecuteBroadcast(BroadcastExchangeExec.scala:146)
at org.apache.spark.sql.execution.InputAdapter.doExecuteBroadcast(WholeStageCodegenExec.scala:387)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeBroadcast.apply(SparkPlan.scala:144)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeBroadcast.apply(SparkPlan.scala:140)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery.apply(SparkPlan.scala:155)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:152)
at org.apache.spark.sql.execution.SparkPlan.executeBroadcast(SparkPlan.scala:140)
at org.apache.spark.sql.execution.joins.BroadcastHashJoinExec.prepareBroadcast(BroadcastHashJoinExec.scala:117)
...
Cause: java.lang.IllegalStateException: Cannot call methods on a stopped SparkContext.
This stopped SparkContext was created at:
org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:926)
com.aravind.oss.SparkSessionTestWrapper$class.spark(SparkSessionTestWrapper.scala:16)
com.aravind.oss.eg.spark.sales.ProductSalesAppTestSpec.spark$lzycompute(ProductSalesAppTestSpec.scala:13)
com.aravind.oss.eg.spark.sales.ProductSalesAppTestSpec.spark(ProductSalesAppTestSpec.scala:13)
com.aravind.oss.eg.spark.sales.ProductSalesAppTestSpec.<init>(ProductSalesAppTestSpec.scala:18)
sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
java.lang.reflect.Constructor.newInstance(Constructor.java:422)
java.lang.Class.newInstance(Class.java:442)
org.scalatest.tools.DiscoverySuite$.getSuiteInstance(DiscoverySuite.scala:66)
org.scalatest.tools.DiscoverySuite$$anonfun.apply(DiscoverySuite.scala:38)
org.scalatest.tools.DiscoverySuite$$anonfun.apply(DiscoverySuite.scala:37)
scala.collection.TraversableLike$$anonfun$map.apply(TraversableLike.scala:234)
scala.collection.TraversableLike$$anonfun$map.apply(TraversableLike.scala:234)
scala.collection.Iterator$class.foreach(Iterator.scala:891)
scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
scala.collection.AbstractIterable.foreach(Iterable.scala:54)
scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
The currently active SparkContext was created at:
org.apache.spark.sql.SparkSession$Builder.getOrCreate(SparkSession.scala:926)
com.aravind.oss.SparkSessionTestWrapper$class.spark(SparkSessionTestWrapper.scala:16)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec.spark$lzycompute(WordCountDSAppTestSpec.scala:15)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec.spark(WordCountDSAppTestSpec.scala:15)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec$$anonfun.apply$mcV$sp(WordCountDSAppTestSpec.scala:20)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec$$anonfun.apply(WordCountDSAppTestSpec.scala:19)
com.aravind.oss.eg.spark.wordcount.WordCountDSAppTestSpec$$anonfun.apply(WordCountDSAppTestSpec.scala:19)
org.scalatest.OutcomeOf$class.outcomeOf(OutcomeOf.scala:85)
org.scalatest.OutcomeOf$.outcomeOf(OutcomeOf.scala:104)
org.scalatest.Transformer.apply(Transformer.scala:22)
org.scalatest.Transformer.apply(Transformer.scala:20)
org.scalatest.FlatSpecLike$$anon.apply(FlatSpecLike.scala:1682)
org.scalatest.TestSuite$class.withFixture(TestSuite.scala:196)
org.scalatest.FlatSpec.withFixture(FlatSpec.scala:1685)
org.scalatest.FlatSpecLike$class.invokeWithFixture(FlatSpecLike.scala:1679)
org.scalatest.FlatSpecLike$$anonfun$runTest.apply(FlatSpecLike.scala:1692)
org.scalatest.FlatSpecLike$$anonfun$runTest.apply(FlatSpecLike.scala:1692)
org.scalatest.SuperEngine.runTestImpl(Engine.scala:286)
org.scalatest.FlatSpecLike$class.runTest(FlatSpecLike.scala:1692)
org.scalatest.FlatSpec.runTest(FlatSpec.scala:1685)
at org.apache.spark.SparkContext.assertNotStopped(SparkContext.scala:100)
at org.apache.spark.SparkContext.defaultParallelism(SparkContext.scala:2359)
at org.apache.spark.sql.execution.LocalTableScanExec.numParallelism$lzycompute(LocalTableScanExec.scala:49)
at org.apache.spark.sql.execution.LocalTableScanExec.numParallelism(LocalTableScanExec.scala:48)
at org.apache.spark.sql.execution.LocalTableScanExec.rdd$lzycompute(LocalTableScanExec.scala:51)
at org.apache.spark.sql.execution.LocalTableScanExec.rdd(LocalTableScanExec.scala:51)
at org.apache.spark.sql.execution.LocalTableScanExec.doExecute(LocalTableScanExec.scala:55)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute.apply(SparkPlan.scala:131)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute.apply(SparkPlan.scala:127)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery.apply(SparkPlan.scala:155)
...
我检查了你的git repo
进入这个测试用例后
"howManyProductsHaveBeenSoldAtLeastOnce" should "return zero when no orders are made for products" in {
当您使用来自 trait 的 spark 会话时，它正在通过 mvn test 使用新线程重置（停止 spark 上下文）您的 spark 会话。
来自 Itellij IDE 它在同一个线程中并且它不重置 spark（不停止上下文）
我能够在 mvn 调试模式 (mvn -DforkMode=never test -f pom.xml) 和使用 spark.sparkContext.isStopped 下进行调查，如下面的代码所示。
这就是它从 intellij 而不是通过 mvn test
所以我放弃了为 spark session 创建 trait 的想法，并创建了一个本地 spark session（也许你可以尝试 import org.scalatest.{BeforeAndAfterAll}）以获得更好的代码。
package com.aravind.oss.eg.spark.wordcount
import com.aravind.oss.SparkSessionTestWrapper
import com.aravind.oss.eg.spark.sales.ProductSalesApp
import com.github.mrpowers.spark.fast.tests.DatasetComparer
import org.scalatest.{BeforeAndAfterAll, FlatSpec}
import ProductSalesApp._
import org.apache.spark.sql.SparkSession
class ProductSalesAppTestSpec extends FlatSpec with DatasetComparer {
lazy implicit val spark: SparkSession =
SparkSession
.builder()
.master("local")
.appName("spark-fast-tests test session")
.config("spark.sql.shuffle.partitions", "1")
.getOrCreate()
import spark.implicits._
val productCols = Seq("product_id", "product_name", "price")
val productDF = Seq(
("0", "product_0", "22"),
("1", "product_1", "30"),
("2", "product_2", "91")
).toDF(productCols: _*)
productDF.createOrReplaceTempView("PRODUCTS")
val orderCols = Seq("order_id", "product_id", "seller_id", "date", "num_pieces_sold", "bill_raw_text")
val orderDF = Seq(
("1", "10", "0", "2020-07-10", "26", "kyeibuumwlyhuwksx"),
("2", "20", "0", "2020-07-08", "13", "kyeibuumwlyhuwksx"),
("3", "30", "0", "2020-07-05", "38", "kyeibuumwlyhuwksx"),
("4", "40", "0", "2020-07-05", "56", "kyeibuumwlyhuwksx")
).toDF(orderCols: _*)
orderDF.createOrReplaceTempView("ORDERS")
orderDF.show
println( spark.sparkContext.isStopped)
println( spark.sparkContext.isLocal)
"howManyProductsHaveBeenSoldAtLeastOnce" should "return zero when no orders are made for products" in {
println( "howManyProductsHaveBeenSoldAtLeastOnce "+ spark.sparkContext.isStopped)
println( spark.sparkContext.isLocal)
println(ProductSalesApp.howManyProductsHaveBeenAoldAtLeastOnce(spark))
assert(ProductSalesApp.howManyProductsHaveBeenAoldAtLeastOnce(spark) == 0)
}
}
结果：
+--------+----------+---------+----------+---------------+-----------------+
|order_id|product_id|seller_id| date|num_pieces_sold| bill_raw_text|
+--------+----------+---------+----------+---------------+-----------------+
| 1| 10| 0|2020-07-10| 26|kyeibuumwlyhuwksx|
| 2| 20| 0|2020-07-08| 13|kyeibuumwlyhuwksx|
| 3| 30| 0|2020-07-05| 38|kyeibuumwlyhuwksx|
| 4| 40| 0|2020-07-05| 56|kyeibuumwlyhuwksx|
+--------+----------+---------+----------+---------------+-----------------+
false
true
howManyProductsHaveBeenSoldAtLeastOnce false
true
0