OrientDB ETL、すでにグラフ内にある2つの頂点の間にエッジを作成

Question

すでにOreintDBの一部である2つの頂点の間にエッジを作成しようとしています。私のエッジデータはMySQLテーブルにあります。

ここは私のoetl jsonです。私は上記のJSONを実行すると

{ 
     "config": { 
     "log": "info" 
     }, 
     "source": { "file": { "path": "/Users/RP/user_invited_data.csv" } }, 
    "extractor": { "csv": {"columnsOnFirstLine": true, "columns":["user_id:string", "invited_by:string", "invited_date:datetime"] } }, 
     "transformers" : [ 
     { "vertex": { "class": "User", "skipDuplicates": true} }, 
     { "edge": { "class": "INVITED", "direction" : "in", 
       "joinFieldName": "invited_by", 
       "lookup":"select expand(u) from (match {class: User, as: u} return u) where u.user_id = ?;", 
       "unresolvedLinkAction":"NOTHING", 
       "edgeFields": { "invited_date": "${input.invited_date}" }, 
       "skipDuplicates": true 
       } 
     }, 
     { "field": 
      { "fieldNames": 
      [ "invited_by", "invited_date"], 
      "operation": "remove" 
      } 
     } 
     ], 
     "loader" : { 
     "orientdb": { 
      "dbURL": "remote:localhost/abcd_graph", 
      "dbUser": "root", 
      "dbPassword": "root", 
      "dbType": "graph", 
      "dbAutoCreate": false, 
      "batchCommit": 1000 
     } 
     } 
    } 

、それはユーザーの頂点のためのORecordDuplicatedExceptionを投げています。 user_idに作成されたユニークなインデックスがあり、skipDuplicates = trueがあります。どんな提案も大歓迎です。

UPDATE：あなたのlogレベルがないDEBUGときOrientDB、skipDuplicatesの 宝石は、実際に動作します。しかし、問題はまだ解決されていません。今はエラーはありませんが、エッジは作成されません。私はそれをデバッグして、今夜それを修正できるかどうかを確認します。

更新日 さらに深くデバッグした後、ストレージレベルが深くなっています。

com.orientechnologies.orient.core.exception.ODatabaseException: Impossible to serialize invalid link #-1:-1 
    DB name="abcd_graph" 
    at com.orientechnologies.orient.core.serialization.serializer.record.binary.ORecordSerializerBinaryV0.writeOptimizedLink(ORecordSerializerBinaryV0.java:867) 
    at com.orientechnologies.orient.core.serialization.serializer.record.binary.ORecordSerializerBinaryV0.serializeValue(ORecordSerializerBinaryV0.java:754) 
    at com.orientechnologies.orient.core.serialization.serializer.record.binary.ORecordSerializerBinaryV0.serialize(ORecordSerializerBinaryV0.java:385) 
    at com.orientechnologies.orient.core.serialization.serializer.record.binary.ORecordSerializerBinary.toStream(ORecordSerializerBinary.java:99) 
    at com.orientechnologies.orient.core.record.impl.ODocument.toStream(ODocument.java:2381) 
    at com.orientechnologies.orient.core.record.impl.ODocument.toStream(ODocument.java:664) 
    at com.orientechnologies.orient.core.db.document.ODatabaseDocumentTx.executeSaveRecord(ODatabaseDocumentTx.java:2183) 
    at com.orientechnologies.orient.core.tx.OTransactionNoTx.saveRecord(OTransactionNoTx.java:191) 
    at com.orientechnologies.orient.core.db.document.ODatabaseDocumentTx.save(ODatabaseDocumentTx.java:2758) 
    at com.orientechnologies.orient.core.db.document.ODatabaseDocumentTx.save(ODatabaseDocumentTx.java:102) 
    at com.orientechnologies.orient.core.record.impl.ODocument.save(ODocument.java:1805) 
    at com.orientechnologies.orient.core.record.impl.ODocument.save(ODocument.java:1801) 
    at com.tinkerpop.blueprints.impls.orient.OrientGraphNoTx.addEdgeInternal(OrientGraphNoTx.java:242) 
    at com.tinkerpop.blueprints.impls.orient.OrientGraphNoTx.addEdgeInternal(OrientGraphNoTx.java:137) 
    at com.tinkerpop.blueprints.impls.orient.OrientVertex.addEdge(OrientVertex.java:741) 
    at com.tinkerpop.blueprints.impls.orient.OrientVertex.addEdge(OrientVertex.java:688) 
    at com.orientechnologies.orient.etl.transformer.OEdgeTransformer.createEdge(OEdgeTransformer.java:203) 
    at com.orientechnologies.orient.etl.transformer.OEdgeTransformer.executeTransform(OEdgeTransformer.java:123) 
    at com.orientechnologies.orient.etl.transformer.OAbstractTransformer.transform(OAbstractTransformer.java:39) 
    at com.orientechnologies.orient.etl.OETLPipeline.execute(OETLPipeline.java:110) 
    at com.orientechnologies.orient.etl.OETLProcessor$OETLPipelineWorker.call(OETLProcessor.java:620) 
    at com.orientechnologies.orient.etl.OETLProcessor$OETLPipelineWorker.call(OETLProcessor.java:601) 
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) 
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) 
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) 
    at java.lang.Thread.run(Thread.java:745) 

UPDATE 再現することが容易になるように、私は、DBからCSVへの抽出を変更しました。

は、スキーマを作成します。

CREATE class User IF NOT EXISTS extends V; 
create property User.user_id IF NOT EXISTS String; 
create property User.name IF NOT EXISTS String; 
create index user_idx on User(user_id) unique; 

insert into User set user_id = '1000_USER1', name = 'Bob'; 
insert into User set user_id = '1001_USER2', name = 'Robert'; 

サンプルCSV：いくつかの闘争の後

user_id, ivited_by, invited_date 
1001_USER2, 1000_USER1, 

Answer 1

と全体のETLのマニュアルおよび一部のデバッグを再読し、私はそれを考え出しました。

VERTEXの代わりにMERGのトランスを使用する必要があります。マージトランスは、作成する代わりにVertexを検索します。ここで

は私のJSONが

"transformers" : [ 
    { "merge": { "joinFieldName": "user_id", "lookup": "User.user_id" } }, 
    { "edge": { "class": "INVITED", "direction" : "out", 
      "joinFieldName": "invited_by", 
      "lookup": "SELECT expand(u) from (match {class: User, as: u} return u) where u.user_id = ?", 
      "unresolvedLinkAction":"NOTHING", 
      "edgeFields": { "invited_date": "${input.invited_date}" }, 
      "skipDuplicates": true 
      } 
    }, 
    { "field": 
     { "fieldNames": 
     [ "invited_by", "invited_date"], 
     "operation": "remove" 
     } 
    } 
    ] 

のように、私はまだお互いの問題を持っていますが、私はそれについて別のものと研究としてそれを取るに見えています。問題は、それは、私は別の問題として、それを取り組む

同じ2つの頂点間の重複エッジを作成されています。

私がOrientDBで常時見てきたことは、そこにあることですが、それを理解することは難しいです。