{"action":"opened","issue":{"url":"https://api.github.com/repos/elastic/elasticsearch-hadoop/issues/745","repository_url":"https://api.github.com/repos/elastic/elasticsearch-hadoop","labels_url":"https://api.github.com/repos/elastic/elasticsearch-hadoop/issues/745/labels{/name}","comments_url":"https://api.github.com/repos/elastic/elasticsearch-hadoop/issues/745/comments","events_url":"https://api.github.com/repos/elastic/elasticsearch-hadoop/issues/745/events","html_url":"https://github.com/elastic/elasticsearch-hadoop/issues/745","id":148236592,"number":745,"title":"Adding  specific Spark partitioner  implematation to speed Bulk  ","user":{"login":"allwefantasy","id":797758,"avatar_url":"https://avatars.githubusercontent.com/u/797758?v=3","gravatar_id":"","url":"https://api.github.com/users/allwefantasy","html_url":"https://github.com/allwefantasy","followers_url":"https://api.github.com/users/allwefantasy/followers","following_url":"https://api.github.com/users/allwefantasy/following{/other_user}","gists_url":"https://api.github.com/users/allwefantasy/gists{/gist_id}","starred_url":"https://api.github.com/users/allwefantasy/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/allwefantasy/subscriptions","organizations_url":"https://api.github.com/users/allwefantasy/orgs","repos_url":"https://api.github.com/users/allwefantasy/repos","events_url":"https://api.github.com/users/allwefantasy/events{/privacy}","received_events_url":"https://api.github.com/users/allwefantasy/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":0,"created_at":"2016-04-14T02:55:26Z","updated_at":"2016-04-14T02:55:26Z","closed_at":null,"body":"\r\n\r\nIt seens if we can partitions data by sharding algrithom from ES in spark, \r\nes-hadoop bulk QPS will be double.\r\n\r\nThis kind of preprocessing have benefit like these:\r\n\r\n* avoid N*N times communication between ES Nodes\r\n* avoid consuming to much bulk thread in ES\r\n\r\nHowever, this means parition num in spark is equal to shard num in ES. So we also can map multi paritions to one shard ,like this:\r\n\r\n```\r\n//variable shardToPartitions means how many paritions are mapped to one partition\r\nclass ESShardPartitioner {\r\n.....\r\noverride def getPartition(key: Any): Int = {\r\n        val shardId = ShardAlg.shard(key.toString(), _numPartitions)        \r\n        val paritions = ((shardId * shardToPartitions) until (shardId * shardToPartitions + shardToPartitions)).toList\r\n        val targetPartitionId = Random.shuffle(paritions).head        \r\n        targetPartitionId\r\n      }\r\n....\r\n}\r\n\r\nobject ESShardPartitioner {\r\n\r\ndef shardIdFromPartitionId(partitionId:Int,shardToPartitions:Int):Int={\r\n     if (partitionId < shardToPartitions) return 0 \r\n      partitionId / shardToPartitions   \r\n   } \r\n   .....\r\n   }     \r\n```\r\n\r\nIn order to  support this, we add a new property:\r\n\r\n```\r\nString  ES_BULK_SHARDID = \"es.bulk.shardid\";\r\n```\r\n\r\nAnd make some changes in RestService.initSingleIndex \r\n\r\n```\r\nif(settings.getProperty(ConfigurationOptions.ES_BULK_SHARDID) != null){       \r\n        \ttargetNode = targetShards.get(orderedShards.get(Integer.parseInt(settings.getProperty(ConfigurationOptions.ES_BULK_SHARDID))));\r\n        }\r\n```\r\n\r\nNow we can use like this:\r\n\r\n```\r\n.mapPartitionsWithIndex { (partionId, iter) =>\r\n    \r\n    val writer = EsSpark.createEsRDDWriter[Map[String,String]](settings)\r\n    val shardId = ESShardPartitioner.shardIdFromPartitionId(partionId, shardToPartitions)\r\n    settings.setProperty(\"es.bulk.shardid\",targetShardId.toString())\r\n    writer.write(TaskContext.get(), shardId, iter.map(f => f._2))\r\n\r\n\r\n.....\r\n}\r\n```\r\n \r\n"}}