Como hacer Bulk Insert con Doctrine DBAL

Los ORM te dan por un lado muchas ventajas, pero por otro lado lo que ganas en sencillez, en abstracción y en encapsulación lo pierdes en rendimiento. Hoy mismo me he encontrado con un problema de rendimiento al realizar un bulk insert dentro del contexto de una Request con Doctrine. Depende tu tipo de proyecto y necesidades, a veces hay que salirse fuera de los mandos del ORM y trabajar directamente contra nuestro sistema de base de datos.

Para ello vamos a conocer más profundo nuestro ORM. Doctrine está dividido en tres principales paquetes:

Common
DBAL (inclue Common)
ORM (incluye Common+DBAL)

El que nos interesa parte este artículo es el paquete DBAL. Esta pieza de la arquitectura contiene una capa de abstracción de acceso a base de datos situada encima de PDO cuyo objetivo es proporcionar una API que haga de puente entre los diferentes sistemas gestores de bases de datos. En la siguiente slide podéis ver los componentes de la arquitectura:

Slide obtenida de la presentación de Juan Ramón Laguardia

La necesidad que me ha surgido era realizar la inserción de una cantidad de registros altas. Buscando distintas opciones para hacer un bulk insert de estas entidades y optimizar estas inserciones me he encontrado con la siguiente clase BulkInsertQuery que me ha resultado de gran utilidad y he reducido el tiempo de ejecución de mi insert masivo a través de la capa de ORM. Espero que os sea util y podáis aprovecharla como yo lo he hecho hoy:

Bulk Insert Query

<?php
namespace Alonsus\Services;
use Doctrine\DBAL\Connection;
use Doctrine\DBAL\Schema\Identifier;
/**
* Class BulkInsertQuery
* @package Alonsus\Services
*/
class BulkInsertQuery
{
/** @var Connection */
protected $connection;
/** @var Identifier */
protected $table;
/** @var string[] */
protected $columns = [];
/** @var array[] */
protected $valueSets = [];
/** @var int[] PDO::PARAM_* */
protected $types = [];
/** @var int|null */
protected $lastInsertId = null;
/** @var int|null */
protected $numInsertedRows = null;
/**
* BulkInsertQuery constructor.
*
* @param Connection $connection
* @param string $table
*/
public function __construct(
Connection $connection,
$table
)
{
$this->connection = $connection;
$this->table = new Identifier($table);
}
/**
* @param array $columns
*
* @return $this
*/
public function setColumns(array $columns)
{
$this->columns = $columns;
return $this;
}
/**
* @param array $valueSets
* @param array|null $types
*
* @return $this
*/
public function setValues(array $valueSets, array $types = null)
{
$this->valueSets = $valueSets;
$this->types = $types;
return $this;
}
/**
* @return $this
*/
public function execute()
{
$sql = $this->getSQL();
$parameters = array_reduce($this->valueSets, function (array $flattenedValues, array $valueSet) {
return array_merge($flattenedValues, array_values($valueSet));
}, []);
$this->connection->executeQuery($sql, $parameters, $this->getPositionalTypes());
$this->lastInsertId = $this->connection->lastInsertId();
$this->numInsertedRows = count($this->valueSets);
return $this;
}
/**
* @return array
*/
public function getLastInsertIds()
{
$lastInsertIds = [];
if (null !== $this->lastInsertId && $this->numInsertedRows > 0) {
$lastInsertIds = range(
$this->lastInsertId,
$this->lastInsertId + $this->numInsertedRows - 1
);
}
return $lastInsertIds;
}
/**
* @return string
*/
protected function getSQL()
{
$platform = $this->connection->getDatabasePlatform();
$escapedColumns = array_map(function ($column) use ($platform) {
return (new Identifier($column))->getQuotedName($platform);
}, $this->columns);
// (id, name, ..., date)
$columnString = empty($this->columns) ? '' : '(' . implode(', ', $escapedColumns) . ')';
// (?, ?, ?, ... , ?)
$singlePlaceholder = '(' . implode(', ', array_fill(0, count($this->columns), '?')) . ')';
// (?, ?), ... , (?, ?)
$placeholders = implode(', ', array_fill(0, count($this->valueSets), $singlePlaceholder));
$sql = sprintf(
'INSERT INTO %s %s VALUES %s;',
$this->table->getQuotedName($platform),
$columnString,
$placeholders
);
return $sql;
}
/**
* @return int[] PDO::PARAM_*
*/
protected function getPositionalTypes()
{
if (empty($this->types)) {
return [];
}
$types = array_values($this->types);
$repeat = count($this->valueSets);
$positionalTypes = [];
for ($i = 1; $i <= $repeat; $i++) {
$positionalTypes = array_merge($positionalTypes, $types);
}
return $positionalTypes;
}
}

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

<?php

namespace Alonsus\Services;

use Doctrine\DBAL\Connection;

use Doctrine\DBAL\Schema\Identifier;

/**

* Class BulkInsertQuery

* @package Alonsus\Services

class BulkInsertQuery

{

/** @var Connection */

protected $connection;

/** @var Identifier */

protected $table;

/** @var string[] */

protected $columns = [];

/** @var array[] */

protected $valueSets = [];

/** @var int[] PDO::PARAM_* */

protected $types = [];

/** @var int|null */

protected $lastInsertId = null;

/** @var int|null */

protected $numInsertedRows = null;

/**

* BulkInsertQuery constructor.

* @param Connection $connection

* @param string $table

public function __construct(

Connection $connection,

$table

)

{

$this->connection = $connection;

$this->table = new Identifier($table);

}

/**

* @param array $columns

* @return $this

public function setColumns(array $columns)

{

$this->columns = $columns;

return $this;

}

/**

* @param array $valueSets

* @param array|null $types

* @return $this

public function setValues(array $valueSets, array $types = null)

{

$this->valueSets = $valueSets;

$this->types = $types;

return $this;

}

/**

* @return $this

public function execute()

{

$sql = $this->getSQL();

$parameters = array_reduce($this->valueSets, function (array $flattenedValues, array $valueSet) {

return array_merge($flattenedValues, array_values($valueSet));

}, []);

$this->connection->executeQuery($sql, $parameters, $this->getPositionalTypes());

$this->lastInsertId = $this->connection->lastInsertId();

$this->numInsertedRows = count($this->valueSets);

return $this;

}

/**

* @return array

public function getLastInsertIds()

{

$lastInsertIds = [];

if (null !== $this->lastInsertId && $this->numInsertedRows > 0) {

$lastInsertIds = range(

$this->lastInsertId,

$this->lastInsertId + $this->numInsertedRows - 1

);

}

return $lastInsertIds;

}

/**

* @return string

protected function getSQL()

{

$platform = $this->connection->getDatabasePlatform();

$escapedColumns = array_map(function ($column) use ($platform) {

return (new Identifier($column))->getQuotedName($platform);

}, $this->columns);

// (id, name, ..., date)

$columnString = empty($this->columns) ? '' : '(' . implode(', ', $escapedColumns) . ')';

// (?, ?, ?, ... , ?)

$singlePlaceholder = '(' . implode(', ', array_fill(0, count($this->columns), '?')) . ')';

// (?, ?), ... , (?, ?)

$placeholders = implode(', ', array_fill(0, count($this->valueSets), $singlePlaceholder));

$sql = sprintf(

'INSERT INTO %s %s VALUES %s;',

$this->table->getQuotedName($platform),

$columnString,

$placeholders

);

return $sql;

}

/**

* @return int[] PDO::PARAM_*

protected function getPositionalTypes()

{

if (empty($this->types)) {

return [];

}

$types = array_values($this->types);

$repeat = count($this->valueSets);

$positionalTypes = [];

for ($i = 1; $i <= $repeat; $i++) {

$positionalTypes = array_merge($positionalTypes, $types);

}

return $positionalTypes;

}

Para utilizar esta clase debemos de instanciarla pasándole la conexión de Doctrine y el nombre de la tabla tal cual se encuentra en nuestro esquema. Se deben rellenar las columnas que queremos llenar en el insert y sus valores. Tras ejecutar serás feliz y verás como los tiempos de inserción se han reducido considerablemente.

$bulkInserQuery = new BulkInsertQuery($this->entityManager->getConnection(), 'table');
$bulkInserQuery->setColumns(array('fecha_inicio', 'documento_id', 'exportacion_id', 'estado'));
$bulkInserQuery->setValues($rows);
$bulkInserQuery->execute();

$bulkInserQuery = new BulkInsertQuery($this->entityManager->getConnection(), 'table');

$bulkInserQuery->setColumns(array('fecha_inicio', 'documento_id', 'exportacion_id', 'estado'));

$bulkInserQuery->setValues($rows);

$bulkInserQuery->execute();

BLOG

Como hacer Bulk Insert con Doctrine DBAL

Bulk Insert Query

Enjoy Progamming 😀

BLOG

Como hacer Bulk Insert con Doctrine DBAL

Bulk Insert Query

Enjoy Progamming 😀

Nube de Tags

Publicaciones Relacionadas

Conociendo una herramienta ORM: Hibernate (II)

Doctrine y las bases de datos NoSQL

Aprende ya que es un ORM