Tutoriel · Salesforce · Bulk API
Bulk API 2.0 avec Symfony : chargement en streaming et reprise
Des centaines de milliers de lignes sans saturer la mémoire : CSV produit en flux, job suivi par un message Messenger différé, reprise des verrous et extraction page par page.
Le principe
L’article Bulk API 2.0 : charger et extraire des millions d’enregistrements décrit le cycle d’un job en cURL. Dans une application Symfony, ce cycle doit tenir trois contraintes : ne jamais charger tout le volume en mémoire, ne pas bloquer un processus pendant que Salesforce traite, et reprendre proprement les lignes refusées.
Fil rouge : chaque nuit, le middleware d’Acme Assurances, compagnie fictive, pousse vers son org acme-demo les clients modifiés dans sa base locale, soit quelques centaines de milliers de lignes certains jours.
app:salesforce:push-clients (commande, planifiée la nuit)
├─ lit la table client par flux, écrit des CSV de 90 Mo maximum
├─ pour chaque CSV : crée un job, envoie, ferme ──► Salesforce traite en arrière-plan
└─ publie PollBulkJob(jobId) avec un délai
PollBulkJob (worker Messenger)
├─ job en cours : se republie 30 s plus tard
└─ job terminé : télécharge les échecs, rejoue les verrous, journalise les rejetsPour quelques dizaines d’écritures au fil de l’eau, une file Messenger et des lots de 200 restent plus simples : la Bulk API a un coût fixe de création et de suivi de job.
Produire le CSV en streaming
Le résultat SQL se lit ligne par ligne avec iterateNumeric() et s’écrit aussitôt sur disque. La mémoire consommée reste constante, qu’il y ait mille ou deux millions de clients. Le fichier change dès qu’il approche la taille maximale d’un envoi.
<?php
namespace App\Salesforce\Bulk;
use Doctrine\DBAL\Connection;
final class ClientCsvExporter
{
private const MAX_BYTES = 90 * 1024 * 1024; // marge sous la limite d'environ 100 Mo par envoi
private const HEADER = ['Ref_Externe__c', 'FirstName', 'LastName', 'Email', 'Account.Ref_Externe__c'];
public function __construct(private readonly Connection $db)
{
}
/** @return \Generator<string> chemins des CSV produits */
public function export(\DateTimeImmutable $since): \Generator
{
// Tri par compte parent : les contacts d'un même compte restent dans le même lot Salesforce (moins de verrous).
$stmt = $this->db->prepare(
'SELECT ref, prenom, nom, email, ref_agence FROM client WHERE modifie_le >= :since ORDER BY ref_agence, ref'
);
$stmt->bindValue('since', $since->format('Y-m-d H:i:s'));
$rows = $stmt->executeQuery()->iterateNumeric(); // ligne par ligne, sans tout charger en mémoire
$file = null;
foreach ($rows as $row) {
if ($file === null || ftell($file) >= self::MAX_BYTES) {
if ($file !== null) {
fclose($file);
yield $path;
}
$path = tempnam(sys_get_temp_dir(), 'bulk-') . '.csv';
$file = fopen($path, 'w');
self::put($file, self::HEADER);
}
// "#N/A" vide un champ dans Salesforce ; une cellule vide le laisse inchangé.
$row[3] ??= '#N/A';
self::put($file, $row);
}
if ($file !== null) {
fclose($file);
yield $path;
}
}
private static function put($file, array $fields): void
{
// escape '' : CSV RFC 4180 (guillemets doublés), le format attendu par la Bulk API.
fputcsv($file, $fields, ',', '"', '', "\n");
}
}Deux détails de format évitent des rejets silencieux : l’échappement à la RFC 4180 (sans antislash, que PHP utilise par défaut), et la valeur #N/A pour vider un champ, car une cellule vide est ignorée.
Un client Bulk API
Le client réutilise le fournisseur de jeton de l’article OAuth. L’envoi passe une ressource de fichier à HttpClient, qui la transmet par morceaux :
<?php
namespace App\Salesforce\Bulk;
use App\Salesforce\SalesforceTokenProvider;
use Symfony\Contracts\HttpClient\HttpClientInterface;
final class BulkIngestClient
{
public function __construct(
private readonly HttpClientInterface $http,
private readonly SalesforceTokenProvider $tokens,
private readonly string $apiVersion = 'v62.0',
) {
}
public function upsert(string $object, string $externalId, string $csvPath): string
{
$job = $this->call('POST', '/jobs/ingest', ['json' => [
'object' => $object,
'operation' => 'upsert',
'externalIdFieldName' => $externalId,
'contentType' => 'CSV',
'lineEnding' => 'LF',
]])->toArray();
// Corps passé en ressource : HttpClient l'envoie par morceaux, le fichier n'est jamais chargé en entier.
$this->call('PUT', '/jobs/ingest/' . $job['id'] . '/batches', [
'headers' => ['Content-Type' => 'text/csv'],
'body' => fopen($csvPath, 'r'),
])->getStatusCode();
$this->call('PATCH', '/jobs/ingest/' . $job['id'], ['json' => ['state' => 'UploadComplete']])->getStatusCode();
return $job['id'];
}
public function status(string $jobId): array
{
return $this->call('GET', '/jobs/ingest/' . $jobId)->toArray();
}
/** Écrit le CSV des lignes en échec dans un fichier local, en flux. */
public function downloadFailures(string $jobId, string $target): void
{
$response = $this->call('GET', '/jobs/ingest/' . $jobId . '/failedResults/');
$out = fopen($target, 'w');
foreach ($this->http->stream($response) as $chunk) {
fwrite($out, $chunk->getContent());
}
fclose($out);
}
private function call(string $method, string $path, array $options = [])
{
$token = $this->tokens->token();
return $this->http->request($method, $token['instance_url'] . '/services/data/' . $this->apiVersion . $path, $options + [
'auth_bearer' => $token['access_token'],
]);
}
}La commande planifiée enchaîne export et envoi, puis confie le suivi à Messenger. Un verrou Symfony Lock empêche deux exécutions simultanées :
#[AsCommand(name: 'app:salesforce:push-clients', description: 'Envoie les clients modifiés vers Salesforce (Bulk API 2.0)')]
final class PushClientsCommand extends Command
{
public function __construct(
private readonly ClientCsvExporter $exporter,
private readonly BulkIngestClient $bulk,
private readonly MessageBusInterface $bus,
private readonly LockFactory $locks,
) {
parent::__construct();
}
protected function execute(InputInterface $input, OutputInterface $output): int
{
// Deux exécutions simultanées chargeraient les mêmes lignes et se disputeraient les verrous.
$lock = $this->locks->createLock('salesforce-push-clients', 7200);
if (!$lock->acquire()) {
$output->writeln('Envoi déjà en cours.');
return Command::SUCCESS;
}
foreach ($this->exporter->export(new \DateTimeImmutable('-1 day')) as $csv) {
$jobId = $this->bulk->upsert('Contact', 'Ref_Externe__c', $csv);
$output->writeln("Job $jobId créé");
$this->bus->dispatch(new PollBulkJob($jobId, $csv), [new DelayStamp(30_000)]);
}
return Command::SUCCESS;
}
}Suivre le job sans bloquer
Une boucle sleep() dans la commande immobiliserait un processus pendant tout le traitement Salesforce, parfois une heure. Un message différé fait mieux : le handler lit l’état, et si le job tourne encore, se republie avec un DelayStamp. Entre deux passages, aucun processus n’attend.
#[AsMessageHandler]
final class PollBulkJobHandler
{
public function __construct(
private readonly BulkIngestClient $bulk,
private readonly BulkFailureHandler $failures,
private readonly MessageBusInterface $bus,
private readonly LoggerInterface $logger,
) {
}
public function __invoke(PollBulkJob $message): void
{
$job = $this->bulk->status($message->jobId);
match ($job['state']) {
'UploadComplete', 'InProgress' => $this->bus->dispatch($message, [new DelayStamp(30_000)]),
'JobComplete' => $this->finish($message, $job),
default => $this->logger->critical('Job Bulk {id} : {state} ({error})', [
'id' => $message->jobId, 'state' => $job['state'], 'error' => $job['errorMessage'] ?? '',
]),
};
}
private function finish(PollBulkJob $message, array $job): void
{
$this->logger->info('Job Bulk {id} terminé : {ok} traités, {ko} en échec', [
'id' => $message->jobId, 'ok' => $job['numberRecordsProcessed'], 'ko' => $job['numberRecordsFailed'],
]);
// JobComplete ne veut pas dire « sans erreur » : lire numberRecordsFailed.
if ($job['numberRecordsFailed'] > 0) {
$this->failures->handle($message);
}
unlink($message->csvPath);
}
}Le message PollBulkJob porte l’identifiant du job, le chemin du CSV et un numéro de tentative. Le DelayStamp exige un transport qui sait différer (Doctrine, Redis, AMQP).
Reprendre les lignes en échec
Le fichier failedResults reprend chaque ligne refusée précédée de son erreur. Deux familles se traitent différemment :
- Verrous (
UNABLE_TO_LOCK_ROW) : passagers, dus aux lots traités en parallèle sur un même compte. Les lignes repartent telles quelles dans un nouveau job, trois fois au plus. - Données (e-mail invalide, champ requis, règle de validation) : définitives. Elles rejoignent une table de rejets consultée par l’équipe métier.
public function handle(PollBulkJob $message): void
{
$report = sys_get_temp_dir() . '/echecs-' . $message->jobId . '.csv';
$this->bulk->downloadFailures($message->jobId, $report);
$in = fopen($report, 'r');
$header = fgetcsv($in, escape: ''); // "sf__Id","sf__Error", puis les colonnes d'origine
$retry = null;
while (($line = fgetcsv($in, escape: '')) !== false) {
$error = $line[1];
$original = array_slice($line, 2);
if (str_starts_with($error, 'UNABLE_TO_LOCK_ROW') && $message->attempt < 3) {
// Erreur passagère : la ligne repart, telle quelle, dans un nouveau job.
$retry ??= self::openCsv(array_slice($header, 2), $retryPath);
fputcsv($retry, $original, escape: '', eol: "\n");
continue;
}
// Erreur de données : à corriger à la source, on la consigne pour l'équipe métier.
$this->rejects->record($message->jobId, $original[0], $error);
}
fclose($in);
if ($retry !== null) {
fclose($retry);
$jobId = $this->bulk->upsert('Contact', 'Ref_Externe__c', $retryPath);
$this->bus->dispatch(new PollBulkJob($jobId, $retryPath, $message->attempt + 1), [new DelayStamp(60_000)]);
}
}Le chargement est un upsert sur External ID : rejouer une ligne déjà acceptée ne crée pas de doublon, ce qui rend la reprise sûre même après un incident au milieu du traitement.
Extraire vers la base locale
Le sens inverse suit le même schéma : la commande crée un job de requête (POST /jobs/query), un message différé attend JobComplete, puis les résultats se lisent page par page, chaînés par l’en-tête Sforce-Locator :
public function pullAccounts(string $jobId): int
{
$locator = null;
$total = 0;
do {
$response = $this->call('GET', '/jobs/query/' . $jobId . '/results', ['query' => array_filter([
'maxRecords' => 50000,
'locator' => $locator,
])]);
$rows = self::parseCsv($response->getContent());
// Une transaction par page : une coupure au milieu ne laisse pas de page à moitié écrite.
$this->db->transactional(function (Connection $db) use ($rows): void {
$stmt = $db->prepare('REPLACE INTO agence_sf (sf_id, ref, nom) VALUES (:id, :ref, :nom)');
foreach ($rows as $row) {
$stmt->bindValue('id', $row['Id']);
$stmt->bindValue('ref', $row['Ref_Externe__c']);
$stmt->bindValue('nom', $row['Name']);
$stmt->executeStatement();
}
});
$total += count($rows);
// La chaîne "null" signale la dernière page.
$locator = $response->getHeaders()['sforce-locator'][0] ?? 'null';
} while ($locator !== 'null');
return $total;
}Avec 50 000 lignes par page, la mémoire reste bornée. Pour une extraction incrémentale, filtrer sur SystemModstamp et conserver la date du dernier passage réussi ; la requête doit rester sélective.
Bonnes pratiques
- Journaliser les identifiants de job : ils permettent de retrouver un chargement dans Setup → Bulk Data Load Jobs et de télécharger ses résultats pendant plusieurs jours.
- Surveiller les quotas : les jobs et les enregistrements traités entrent dans des limites par 24 heures, visibles via la ressource
/limits. - Tester en volume : triggers, flows et règles de validation s’exécutent pour chaque ligne ; une sandbox complète révèle les automatisations trop lentes avant la production.
- Nettoyer : supprimer les CSV temporaires une fois le job terminé ; ils contiennent des données personnelles.
- Interrompre proprement : un job ouvert par erreur se ferme avec l’état
Abortedplutôt que d’être abandonné.