nebula:3.0.0
spark:2.2.0
nebula配置:3台虚拟机 8C 16G
目前由于spark版本受限用不了spark-nebula这个包,参考官方nebula-spark-exchange代码,通过spark 的ForeachPartitionFunction进行数据写入,但是经常会报错:All servers are broken
部分代码如下:
@Override
public void call(Iterator<Row> iterator) {
// colNameAll务必要放到call方法里面,否则容器字段值不会被更新
List<String> colNameAll = new LinkedList<>();
// colNameAll务必要放到call方法里面,否则容器字段值不会被更新
List<String> colTypes = new LinkedList<>();
GraphProvider graphProvider = new GraphProvider();
Session session;
try {
session = graphProvider.getSession();
} catch (Exception e) {
// 防止取不到session,休眠10s
try {
Thread.sleep(10000);
} catch (Exception interruptedException) {
interruptedException.printStackTrace();
}
try {
// 再次提取session
session = graphProvider.getSession();
} catch (Exception e1) {
throw new RuntimeException(" get session error : " + e1.getMessage());
}
}
...
graphProvider.releaseSession(session);
graphProvider.close();