通过kafka-nebula的过程出现边数据丢失,传130多万只有100多万左右边成功插入,exchange有读到130多万条数据并处理。但是通过maxcompute-nebula边能全部传入,无重复rank值,请问有没有什么办法解决,这是什么原因导致的问题呢?
kafka-nebula配置信息:
{
# Spark 相关配置
spark: {
app: {
name: Nebula Exchange 3.0.0
}
driver: {
cores: 1
maxResultSize: 0
}
cores: {
max: 16
}
}
# Nebula Graph 相关配置
nebula: {
address:{
# 以下为 Nebula Graph 的 Graph 服务和 Meta 服务所在机器的 IP 地址及端口。
# 如果有多个地址,格式为 "ip1:port","ip2:port","ip3:port"。
# 不同地址之间以英文逗号 (,) 隔开。
graph:["ip:9669"]
meta:["ip:9559"]
}
# 填写的账号必须拥有 Nebula Graph 相应图空间的写数据权限。
user: root
pswd: ***
# 填写 Nebula Graph 中需要写入数据的图空间名称。
space: graphchain_prod
connection: {
timeout: 3000
retry: 3
}
execution: {
retry: 3
}
error: {
max: 32
output: /tmp/errors
}
rate: {
limit: 1024
timeout: 1000
}
}
# 处理边数据
edges: [
# 设置 Edge type common_transaction 相关信息
{
# Nebula Graph 中对应的 Edge type 名称。
name: common_transaction
type: {
# 指定数据源文件格式,设置为 Kafka。
source: kafka
# 指定边数据导入 Nebula Graph 的方式,
# 指定如何将点数据导入 Nebula Graph:Client 或 SST。
sink: client
}
# Kafka 服务器地址。
service: "***"
# 消息类别。
topic: "***"
# 在 fields 里指定 Kafka value 中的字段名称,多个字段用英文逗号(,)隔开。Spark Structured Streaming 读取 Kafka 数据后会将其以 JSON 格式存储于 value 字段中,而这里的 fields 要配置 JSON 的 key 名。示例如下:
fields: [hash,from_address,to_address,block_hash,block_no,currency,currency_name,currency_symbol,quantity,circa_quantity,fee,tx_time,blockchain,tx_index]
# 设置与 fields 中的 key 对应的 Nebula Graph 属性名,key 的 value 将保存为相应的属性值。下方设置会将 degree 的 value 保存到 Nebula Graph 中的 degree 属性。
nebula.fields: [hash,from_address,to_address,block_hash,block_no,currency,currency_name,currency_symbol,quantity,circa_quantity,fee,tx_time,blockchain,tx_index]
# 在 source 里,将 topic 中某一列作为边的起始点数据源。
# 在 target 里,将 topic 中某一列作为边的目的点数据源。
source:{
field:from_vid
}
target:{
field:to_vid
}
# 指定一个列作为 rank 的源(可选)。
#ranking: rank
# 单批次写入 Nebula Graph 的数据条数。
batch: 100
# Spark 分区数量
partition: 32
# 读取消息的间隔。单位:秒。
interval.seconds: 600
}
]
}
运行日志:
"stateOperators" : [ ],
"sources" : [ {
"description" : "KafkaV2[Subscribe[eth_common_transaction]]",
"startOffset" : {
"eth_common_transaction" : {
"8" : 4011989,
"11" : 4011989,
"2" : 4011990,
"5" : 4011987,
"4" : 4011989,
"7" : 4011987,
"10" : 4011986,
"1" : 4011989,
"9" : 4011990,
"3" : 4011988,
"6" : 4011987,
"0" : 4011990
}
},
"endOffset" : {
"eth_common_transaction" : {
"8" : 4122239,
"11" : 4122239,
"2" : 4122240,
"5" : 4122236,
"4" : 4122256,
"7" : 4122254,
"10" : 4122253,
"1" : 4122256,
"9" : 4122326,
"3" : 4122324,
"6" : 4122323,
"0" : 4122326
}
},
"numInputRows" : 1323411,
"inputRowsPerSecond" : 13234.11,
"processedRowsPerSecond" : 6937.425300371139