WebHdfsPersistReader ElasticsearchPersistWriter String es://{index}/{type} hdfs://{indexes}/{types}