src/Commands/ReindexCommand.php
function migrate($original_index, $new_index, $scroll_id = null, $errors = 0, $page = 1)
    {

        if ($page == 1) {

            $pages = (int)ceil($this->es->connection($this->connection)->index($original_index)->count() / $this->size);

            $this->output->progressStart($pages);

            $documents = $this->es->connection($this->connection)->index($original_index)->type("")
                ->scroll($this->scroll)
                ->take($this->size)
                ->response();

        } else {

            $documents = $this->es->connection($this->connection)->index($original_index)->type("")
                ->scroll($this->scroll)
                ->scrollID($scroll_id)
                ->response();

        }

        if (isset($documents["hits"]["hits"]) and count($documents["hits"]["hits"])) {

            $data = $documents["hits"]["hits"];

            $params = [];

            foreach ($data as $row) {

                $params["body"][] = [

                    'index' => [
                        '_index' => $new_index,
                        '_type' => $row["_type"],
                        '_id' => $row["_id"]
                    ]

                ];

                $params["body"][] = $row["_source"];

            }

            $response = $this->es->connection($this->connection)->raw()->bulk($params);

            if (isset($response["errors"]) and $response["errors"]) {

                if (!$this->option("hide-errors")) {

                    if ($this->option("skip-errors")) {
                        $this->warn("\n" . json_encode($response["items"]));
                    } else {
                        return $this->warn("\n" . json_encode($response["items"]));
                    }

                }

                $errors++;
            }

            $this->output->progressAdvance();

        } else {

            // Reindexing finished

            $this->output->progressFinish();

            $total = $this->es->connection($this->connection)->index($original_index)->count();

            if ($errors > 0) {
                return $this->warn("$total documents reindexed with $errors errors.");
            } else {
                return $this->info("$total documents reindexed $errors errors.");
            }

        }

        $page++;

        $this->migrate($original_index, $new_index, $documents["_scroll_id"], $errors, $page);
    }
function migrate($original_index, $new_index, $scroll_id = null, $errors = 0, $page = 1)
    {

        if ($page == 1) {

            $pages = (int)ceil($this->es->connection($this->connection)->index($original_index)->count() / $this->size);

            $this->output->progressStart($pages);

            $documents = $this->es->connection($this->connection)->index($original_index)->type("")
                ->scroll($this->scroll)
                ->take($this->size)
                ->response();

        } else {

            $documents = $this->es->connection($this->connection)->index($original_index)->type("")
                ->scroll($this->scroll)
                ->scrollID($scroll_id)
                ->response();

        }

        if (isset($documents["hits"]["hits"]) and count($documents["hits"]["hits"])) {

            $data = $documents["hits"]["hits"];

            $params = [];

            foreach ($data as $row) {

                $params["body"][] = [

                    'index' => [
                        '_index' => $new_index,
                        '_type' => $row["_type"],
                        '_id' => $row["_id"]
                    ]

                ];

                $params["body"][] = $row["_source"];

            }

            $response = $this->es->connection($this->connection)->raw()->bulk($params);

            if (isset($response["errors"]) and $response["errors"]) {

                if (!$this->option("hide-errors")) {

                    if ($this->option("skip-errors")) {
                        $this->warn("\n" . json_encode($response["items"]));
                    } else {
                        return $this->warn("\n" . json_encode($response["items"]));
                    }

                }

                $errors++;
            }

            $this->output->progressAdvance();

        } else {

            // Reindexing finished

            $this->output->progressFinish();

            $total = $this->es->connection($this->connection)->index($original_index)->count();

            if ($errors > 0) {
                return $this->warn("$total documents reindexed with $errors errors.");
            } else {
                return $this->info("$total documents reindexed $errors errors.");
            }

        }

        $page++;

        $this->migrate($original_index, $new_index, $documents["_scroll_id"], $errors, $page);
    }
function migrate ( $ original_index , $ new_index , $ scroll_id = null , $ errors = 0 , $ page = 1 ) { if ( $ page == 1 ) { $ pages = ( int ) ceil ( $ this -> es -> connection ( $ this -> connection ) -> index ( $ original_index ) -> count ( ) / $ this -> size ) ; $ this -> output -> progressStart ( $ pages ) ; $ documents = $ this -> es -> connection ( $ this -> connection ) -> index ( $ original_index ) -> type ( "" ) -> scroll ( $ this -> scroll ) -> take ( $ this -> size ) -> response ( ) ; } else { $ documents = $ this -> es -> connection ( $ this -> connection ) -> index ( $ original_index ) -> type ( "" ) -> scroll ( $ this -> scroll ) -> scrollID ( $ scroll_id ) -> response ( ) ; } if ( isset ( $ documents [ "hits" ] [ "hits" ] ) and count ( $ documents [ "hits" ] [ "hits" ] ) ) { $ data = $ documents [ "hits" ] [ "hits" ] ; $ params = [ ] ; foreach ( $ data as $ row ) { $ params [ "body" ] [ ] = [ 'index' => [ '_index' => $ new_index , '_type' => $ row [ "_type" ] , '_id' => $ row [ "_id" ] ] ] ; $ params [ "body" ] [ ] = $ row [ "_source" ] ; } $ response = $ this -> es -> connection ( $ this -> connection ) -> raw ( ) -> bulk ( $ params ) ; if ( isset ( $ response [ "errors" ] ) and $ response [ "errors" ] ) { if ( ! $ this -> option ( "hide-errors" ) ) { if ( $ this -> option ( "skip-errors" ) ) { $ this -> warn ( "\n" . json_encode ( $ response [ "items" ] ) ) ; } else { return $ this -> warn ( "\n" . json_encode ( $ response [ "items" ] ) ) ; } } $ errors ++ ; } $ this -> output -> progressAdvance ( ) ; } else { // Reindexing finished $ this -> output -> progressFinish ( ) ; $ total = $ this -> es -> connection ( $ this -> connection ) -> index ( $ original_index ) -> count ( ) ; if ( $ errors > 0 ) { return $ this -> warn ( "$total documents reindexed with $errors errors." ) ; } else { return $ this -> info ( "$total documents reindexed $errors errors." ) ; } } $ page ++ ; $ this -> migrate ( $ original_index , $ new_index , $ documents [ "_scroll_id" ] , $ errors , $ page ) ; }
Migrate data with Scroll queries & Bulk API
@param $original_index
@param $new_index
@param null $scroll_id
@param int $errors
@param int $page
Migrate data with Scroll queries & Bulk API