我有一个 Scala 火花作业,它从 HBase 读取,如下所示:
val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result])
val uniqueAttrs = calculateFreqLocation(hBaseRDD)
我试图为函数calculateFreqLocation编写一个单元测试:
def calculateFreqLocation(inputRDD: RDD[(ImmutableBytesWritable, Result)]): Map[String, Map[(String, String, String), Long]] = {
val valueType = classOf[Array[Attribute]]
val family = "cf_attributes".getBytes()
val qualifier = "attributes".getBytes()
val rdd7 = inputRDD.map(kv => (getUUID(kv._1.get()).toString(),
objectMapper.readValue(new String(kv._2.getValue(family, qualifier)), valueType))).flatMap(flattenRow).filter(t => location_attributes.contains(t._2))
val countByUUID = rdd7.countByValue().groupBy(_._1._1)
val countByUUIDandKey = countByUUID.map(kv => (kv._1, kv._2.groupBy(_._1._2)))
val uniqueAttrs = countByUUIDandKey.map(uuidmap => (uuidmap._1,uuidmap._2.map(keymap => keymap._2.maxBy(_._2))))
return uniqueAttrs
}
这将计算每个 UUID 的唯一属性。我的单元测试尝试重新创建 HTable 数据,然后将 RDD 传递给函数以查看输出是否匹配:
@RunWith(classOf[JUnitRunner])
class FrequentLocationTest extends SparkJobSpec {
"Frequent Location calculation" should {
def longToBytes(x: Long): Array[Byte] = {
return ByteBuffer.allocate(java.lang.Long.SIZE / java.lang.Byte.SIZE).putLong(x).array
}
val currTimestamp = System.currentTimeMillis / 1000
val UUID_1 = UUID.fromString("123456aa-8f07-4190-8c40-c7e78b91a646")
val family = "cf_attributes".getBytes()
val column = "attributes".getBytes()
val row = "[{'name':'Current_Location_Ip_Address', 'value':'123.456.123.248'}]"
val resultRow = Array(new KeyValue(row.getBytes(), family, column, null))
val key = "851971aa-8f07-4190-8c40-c7e78b91a646".getBytes() ++ longToBytes(currTimestamp)
val input = Seq((key,row))
val correctOutput = Map(
("851971aa-8f07-4190-8c40-c7e78b91a646" -> Map(("123456aa-8f07-4190-8c40-c7e78b91a646","Current_Location_Ip_Address","123.456.123.248") -> 1))
)
"case 1 : return with correct output (frequent location calculation)" in {
val inputRDD = sc.makeRDD(input, 1)
val hadoonRdd = new HadoopRDD(sc, sc.broadcast(new SerializableWritable(new Configuration()))
.asInstanceOf[Broadcast[SerializableWritable[Configuration]]], null, classOf[InputFormat[ImmutableBytesWritable,Result]], classOf[ImmutableBytesWritable],classOf[Result],1)
val finalInputRdd = hadoonRdd.union(inputRDD.map(kv => ( new ImmutableBytesWritable(kv._1), new Result(Array(new KeyValue(kv._2.getBytes(), family, column, null))))))
val resultMap = FrequentLocation.calculateFreqLocation(finalInputRdd)
resultMap == correctOutput
//val customCorr = new FrequentLocation().calculateFreqLocation(inputRDD)
//freqLocationMap must_== correctOutput
}
}
}
我得到的是org . Apache . spark . spark exception:任务不可序列化。我开始明白这是因为LongByteWritable和其他HTable类导致spark无法在节点之间序列化。我提供的代码实际上是开发人员的Spark apis(手动创建HadoopRDD ),但没有任何方法实际填充数据。我如何测试这个?我需要将包含数据的HadoopRDD实例返回给这个函数。或RDD的实例(ImmutableBytesWritable,Result)。我最初是手动创建这个RDD,同样的错误。然后我转而使用map并从原始二进制/文本映射它。任何帮助都将不胜感激!
用我自己的发现来回答,为其他也陷入类似堆栈的人提供一些指导:火花在HBase上运行。
如果您学习了大多数单元测试Spark过程的教程,您可能会有这样一个类:
abstract class SparkJobSpec extends SpecificationWithJUnit with BeforeAfterExample {
@transient var sc: SparkContext = _
def beforeAll = {
System.clearProperty("spark.driver.port")
System.clearProperty("spark.hostPort")
val conf = new SparkConf()
.setMaster("local")
.setAppName("test")
//this kryo stuff is of utter importance
.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
.registerKryoClasses(Array(classOf[org.apache.hadoop.hbase.client.Result],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable]))
//.setJars(Seq(System.getenv("JARS")))
sc = new SparkContext(conf)
}
def afterAll = {
if (sc != null) {
sc.stop()
sc = null
System.clearProperty("spark.driver.port")
System.clearProperty("spark.hostPort")
}
}
def before = {}
def after = {}
override def map(fs: => Fragments) = Step(beforeAll) ^ super.map(fs) ^ Step(afterAll)
}
我发布的问题的解决方案实际上是两部分:
>
任务不可序列化异常可以通过将带有可序列化的(如下所示)删除到您的单元测试套件类以及原始Spark过程中轻松修复。显然,在类之间传递RDD实际上序列化了整个类或其他东西?我不知道,但它有所帮助。
我遇到的最大问题是
sparkcontext.newAPIHadoopRDD()
方法非常好,但返回了一个非常奇怪的RDD,形式为RDD(ImmutableBytesWritable,Result)。
两者都不是可序列化的,当你使用这个自我构造的RDD从Spark作业调用函数时,它真的会抱怨这一点。这里的关键是:.set(“spark.serializer”, “org.apache.spark.serializer.KryoSerializer”) .registerKryoClasses
(Array(classOf[org.apache.hadoop.hbase.client.Result],classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable])))设置
在你的sparkConf上。出于某种原因,我不需要在原始的 Spark 过程中执行此操作。不确定这是因为 Spark 在我的 qa 集群中自行执行某些操作,或者我从未在程序之外传递此 RDD,因此它永远不必序列化。
实际上,单元测试的最终代码看起来非常简单!
@RunWith(classOf[JUnitRunner])
class FrequentLocationTest extends SparkJobSpec with Serializable {
"Frequent Location calculation" should {
//some UUID generator stuff here
val resultRow = Array(new KeyValue(Bytes.add(longToBytes(UUID_1.getMostSignificantBits()), longToBytes(UUID_1.getLeastSignificantBits())), family, column, row.getBytes()))
val input = Seq((new ImmutableBytesWritable(key), new Result(resultRow)))
val correctOutput = Map(
("851971aa-8f07-4190-8c40-c7e78b91a646" -> Map(("851971aa-8f07-4190-8c40-c7e78b91a646","Current_Location_Ip_Address","123.456.234.456") -> 1))
)
"case 1 : return with correct output (frequent location calculation)" in {
val inputRDD = sc.makeRDD(input, 1)
val resultMap = FrequentLocation.calculateFreqLocation(inputRDD)
resultMap == correctOutput
}
}
}
问题内容: 我正在尝试为一些依赖WifiManager和返回的ScanResults的类实现一些单元测试。我想做的是能够控制我收到的ScanResults,以测试各种不同的条件。 不幸的是,对我来说,成功模拟WifiManager非常困难(尽管我想我可以在MockWifiManager中传递其构造函数null引用)。这只是我的第一个问题,因为一旦我有一个MockWifiManager可以玩(如果它
我有一个示例方法(我需要编写测试用例)如下所示, 我想模拟getConfig方法并返回一个特定的字符串值。getConfig是Kotlin对象中方法,如下所示, 下面是我尝试的测试 我没有得到任何错误,但是getConfig方法没有被嘲笑。执行实际的实现。我也试过使用Powermockito。请帮帮我
如果我错了请纠正我,但看起来Moq只能模拟一个公共类,它有一个公共的无参数构造函数,要模拟的方法是。我并不想让这些类公开可见。我是不是错过了一些与Moq的东西,或者只是不适合我想做的事情? 我想我可以创建一个ClassB实现的接口(例如“ICLASSB”),将其注入ClassA,并模拟该接口。ClassB仍然可以是内部的(尽管我意识到接口方法必须是公共的)。虽然这可以工作,但我对创建大量接口感到不
Spock对存根和模拟做了很强的区分。当要更改的内容从被测试类使用的类返回时,请使用存根,这样您就可以测试if语句的另一个分支。使用mock,当您不关心测试中的类返回什么时,只需调用另一个类的另一个方法,并且您希望确保调用了该方法。很整洁。然而,假设您有一个具有流利API的构建器,它使人们。您希望测试调用此生成器的方法。 所以最初,我想只是模拟构建器,然后myMethod()的单元测试应该检查具有
由于一个人在单元测试和模拟之前没有经验,我使用Mockito学习了关于JUnit的初学者教程,并进行了一些实践。 现在,我需要对一些在MySQL数据库上执行基本数据库操作的类方法进行单元测试。我不想对数据库进行真正的更改。 在不访问真实数据库的情况下,是否存在验证类似方法的方法?还是我误解了单元测试和嘲笑的概念?
我正在尝试对服务方法进行单元测试。服务方法调用spring数据存储库方法来获取一些数据。我想模拟这个存储库调用,并自己提供数据。如何做到这一点?在Spring Boot文档之后,当我模拟存储库并在测试代码中直接调用存储库方法时,模拟工作正常。但是,当我调用服务方法时,反过来调用存储库方法,mocking就不起作用了。下面是示例代码: 服务级别: 测试等级:
问题内容: 嗨,我真的希望您能为我提供帮助,我感觉我已经把头发拉了好几天了。 我正在尝试为方法A编写单元测试。方法A调用静态方法B。我想模拟静态方法B。 我知道以前已经有人问过这个问题,但是我觉得Android从那时起已经成熟,并且必须有一种方法可以执行如此简单的任务,而无需重写我要测试的方法。 这是一个示例,首先是我要测试的方法: 接下来,我要模拟的静态方法: 在其他语言中,这是如此简单,但我无
问题内容: 我在解决如何使用Apache Mina时遇到了一些麻烦。他们的文档对于我那才华横溢的大脑来说有点乏味。我在Java SFTP服务器库中看到了有用的起始代码 吗? 我不知道如何使用它。我想使用Mina作为一种模拟服务器来设置检查我的sftp代码的单元测试,即能够编写如下单元测试: 问题是要放入什么。 我一直在浏览测试代码,想知道是否需要在上面进行更多配置以指定根目录,用户名和身份验证密钥