ElasticsearchPersistReader WebHdfsPersistWriter String hdfs://{index}/{type} es://{indexes}/{types}