pyspark - 如何防止泰坦生成具有相同属性的重复记录？

Question

我有 spark 应用程序，它使用 goblin 将数据插入到 Titan 中。但它会插入具有相同名称的重复顶点。测试条件“如果不是结果：”不匹配，我在同一个会话中。

def savePartition(p):
    print ('savePartition', p)
    from goblin import element, properties

    class Brand(element.Vertex):
        name = properties.Property(properties.String)

    import asyncio

    loop = asyncio.get_event_loop()

    from goblin.app import Goblin
    app = loop.run_until_complete(Goblin.open(loop))
    app.register(Brand)

    async def go(app):
        session = await app.session()

        for i in p:
            if i['brand']:
                traversal = session.traversal(Brand)
                result = await traversal.has(Brand.name, i['brand']).oneOrNone()

                if not result:  # TODO: Remove Duplicates
                    print(i)
                    brand = Brand()
                    brand.name = i['brand']
                    session.add(brand)
                    session.flush()

        await app.close()

    loop.run_until_complete(go(app))

rdd = rdd.foreachPartition(savePartition)

如何解决？多谢。

Answer 1

我不确定这将如何与 Goblin 一起使用，但如果您希望 Titan 基于顶点属性防止重复，您可以使用 Titan复合索引并指定它们必须是唯一的。例如，您可以执行以下操作：

mgmt = graph.openManagement()
name = mgmt.makePropertyKey('name').dataType(String.class).make()
mgmt.buildIndex('byNameUnique', Vertex.class).addKey(name).unique().buildCompositeIndex()
mgmt.commit()

上面将指定name顶点上的属性必须是唯一的。