如何使用 spark 集群构建 lucene 索引答案

【问题标题】：how to build lucene index with spark cluster如何使用 spark 集群构建 lucene 索引
【发布时间】：2016-12-30 03:50:18
【问题描述】：

我想要一个代码演示或一些想法来使用 spark 集群构建 lucene 索引。我已经尝试了一些方法。但我仍然不知道如何在 spark 中使用 lucene 的 IndexWriter。我的输入数据是这样的：sellerId, productId, title

我希望输出的是 lucene 的索引文件。

【问题讨论】：

标签： apache-spark lucene

【解决方案1】：

你可以看看spark-lucenerdd。有关该库的快速介绍，请查看slides。

免责声明：我是图书馆的作者。

【讨论】：

【解决方案2】：

我实际上做了以下事情，效果很好：

public class LuceneDatasetIndexService implements DatasetIndexService, Serializable
{

    private static final String SCHEMA_JSON = "schema.json";

    @Autowired
    private transient FileUtility fileUtility;

    /**
     * 
     */
    private static final long serialVersionUID = 1L;

    @Override
    public void indexDataset(Dataset<Row> dataset, Path indexStorePath) throws IOException
    {
        if (!fileUtility.exists(indexStorePath))
        {
            Files.createDirectories(indexStorePath);
            Path schemaPath = indexStorePath.resolve(SCHEMA_JSON);
            String prettyJson = dataset.schema().prettyJson();
            Files.copy(new ByteArrayInputStream(prettyJson.getBytes()), schemaPath, StandardCopyOption.REPLACE_EXISTING);
        }
        //Path is not serializable
        String path = indexStorePath.toString();
        dataset.foreachPartition(new ForeachPartitionFunction<Row>()
        {
            /**
             * 
             */
            private static final long serialVersionUID = 1L;

            @Override
            public void call(Iterator<Row> t) throws Exception
            {
                Path indexPath = Paths.get(path);
                StructType schema = SparkSchemaUtil.readSparkSchemaFromFile(indexPath.resolve(SCHEMA_JSON));
                RowIndexWriter writer = null;
                while (writer == null)
                {
                    try
                    {
                        writer = new LuceneRowIndexWriter(schema, indexPath);
                    } catch (LockObtainFailedException e)
                    {
                        Thread.sleep(100);
                    }
                }

                try {
                    while (t.hasNext())
                    {
                        writer.indexRow(t.next());
                    }
                }
                finally 
                {
                    writer.close();
                }
            }
        });
    }

    @Override
    public List<Row> find(Path indexStorePath, Collection<Filter> filters, int size) throws IOException
    {
        Path schemaPath = indexStorePath.resolve(SCHEMA_JSON);

        StructType schema = SparkSchemaUtil.readSparkSchemaFromFile(schemaPath);

        RowIndexSearcher searcher = new LuceneRowIndexSearcher(schema, indexStorePath);
        return searcher.search(filters, indexStorePath, size);
    }

这种方式在索引节点之间没有数据传输，每个分区一次写入一个lucene索引（不能同时打开多个索引器）

【讨论】：