请问是否能提供分布式数据生成的用例？

我现在希望从Edge转化到Block，因为我本身也是从分布式的Edge文件转的（而不是JSON文件），我尝试着用Spark Scala写了个demo，但是输出的文件非常小（只有M级别），请问是哪里出了问题么？同样的伪代码在Python本地是work的。

```
val blockRDD = nEdges.map(r => (r.srcId, Set((r.dstId.toLong, r.attr.toFloat)))).reduceByKey(_ ++ _, 1000)
.map { case (nodeId, neighbors: Set[(Long, Float)]) => {
val block = new Block
block.setNode_id(nodeId)
block.setNode_weight(1.0f)
block.setNodeType(0)
val neighbors_ = neighbors.toArray
val neighbor = new java.util.HashMap[Integer, java.util.HashMap[java.lang.Long, java.lang.Float]]()
for (i <- 0 to meta.getEdge_type_num - 1) {
neighbor.put(i, new java.util.HashMap[java.lang.Long, java.lang.Float])
}
val edgeList = new java.util.ArrayList[EdgeItem]()
val parser = new BlockParser(meta)
for (i <- 0 to neighbors_.length - 1) {
val dstId = neighbors_(i)._1
val weight = neighbors_(i)._2
neighbor.get(0).put(dstId, weight)
val edge = new EdgeItem
edge.setSrc_id(nodeId)
edge.setDst_id(dstId)
edge.setWeight(weight)
edgeList.add(edge)
}
block.setNeighbor(neighbor)
block.setEdge(edgeList)
parser.BlockJsonToBytes(block)
}
}
```

Start with the Spark Scala conversion shown in the issue, especially the reduceByKey pipeline and BlockParser.BlockJsonToBytes call. Compare it with the working Python version and investigate why the generated output is only megabytes. Done means identifying the cause and documenting a reproducible distributed data-generation example.

Written by the indexing model from the issue text.