OraclePropertyGraphDataLoader 从 HDFS 加载数据

Question

我正在使用 Spark+Hive 构建图形和关系，并将平面 OPV/OPE 文件导出到 HDFS，每个 reducer 一个 OPV/OPE CSV。我们所有的图形数据库都已准备好加载到 OPG/PGX 上进行分析，效果非常好。

现在，我们要在 Oracle Property Graph 上加载这些顶点/边。

我通过这种方式从 hdfs 转储文件名:

$ hadoop fs -find '/user/felipeferreira/dadossinapse/ops/*.opv/*.csv' | xargs -I{}  echo 'hdfs://'{} > opvs.lst
$ hadoop fs -find '/user/felipeferreira/dadossinapse/ops/*.ope/*.csv' | xargs -I{}  echo 'hdfs://'{} > opes.lst

我正在 groovy shell 上进行实验，但遇到了一些问题和疑问:

opvs = new File('opvs.lst') as String[]
opes = new File('opes.lst') as String[]

opgdl.loadData(opg, opvs, opes, 72)

这不是开箱即用的，我收到类似的错误

java.lang.IllegalArgumentException: loadData: part-00000-f97f1abf-5f69-479a-baee-ce0a7bcaa86c-c000.csv flat file does not exist

我将使用 loadData 接口(interface)中提供的 InputStream 方法来管理该问题，希望能够解决这个问题，但我有一些问题/建议:

• loadData 是否支持 vfs，以便我可以直接加载“hdfs://...”文件？
• 文件名中包含 glob 语法不太好，因此我们可以这样做:

opgdl.loadData(opg, 'hdfs:///user/felipeferreira/opvs/**/*.csv' ...

提前致谢!

Answer 1

您可以使用 OraclePropertyGraphDataLoader 中的备用 API，您可以在其中为用于加载的 opv/ope 文件指定 InputStream 对象。这样，您就可以使用 FsDataInputStream 对象从 HDFS 环境中读取文件。

一个小样本如下:

// ====== Init HDFS File System Object
Configuration conf = new Configuration();
// Set FileSystem URI
conf.set("fs.defaultFS", hdfsuri);
conf.set("fs.hdfs.impl", org.apache.hadoop.hdfs.DistributedFileSystem.class.getName());
conf.set("fs.file.impl", org.apache.hadoop.fs.LocalFileSystem.class.getName());
// Set HADOOP user
System.setProperty("HADOOP_USER_NAME", "hdfs");
System.setProperty("hadoop.home.dir", "/");

//Get the filesystem - HDFS
FileSystem fs = FileSystem.get(URI.create(hdfsuri), conf);`

// Read files into InputStreams using HDFS FsDataInputStream Java APIs
**Path pathOPV = new Path("/path/to/file.opv");
FSDataInputStream inOPV = fileSystem.open(pathOPV);
Path pathOPV = new Path("/path/to/file.ope");
FSDataInputStream inOPE = fileSystem.open(pathOPE);**

cfg = GraphConfigBuilder.forPropertyGraphHbase().setName("sinapse").setZkQuorum("bda1node05,bda1node06").build()

opg = OraclePropertyGraph.getInstance(cfg)
opgdl = OraclePropertyGraphDataLoader.getInstance();
opgdl.loadData(opg, **inOPV, inOPE**, 100);

请告诉我们这是否适合您。