MuleSoft Mastery: From Zero to Hero | Ch.11

<flow name="enrich-customers">
  <file:listener config-ref="File" directory="/in">
    <scheduling-strategy><fixed-frequency frequency="60000"/></scheduling-strategy>
  </file:listener>

  <!-- payload is now a List<Customer>; the job splits it into records -->
  <batch:job jobName="customerEnrichment" blockSize="100">
    <batch:process-records>

      <!-- only geocode US customers; others skip this step -->
      <batch:step name="geocode" acceptExpression="#[payload.country == 'US']">
        <http:request config-ref="GeoApi" path="/geocode" method="GET">
          <http:query-params>#[{ q: payload.address ++ ', ' ++ payload.city }]</http:query-params>
        </http:request>
        <ee:transform><ee:message><ee:set-payload><![CDATA[%dw 2.0
output application/json
---
payload ++ { lat: payload.lat, lon: payload.lon }]]></ee:set-payload></ee:message></ee:transform>
      </batch:step>

      <!-- bulk-insert accepted records via the aggregator (NOT on-complete) -->
      <batch:step name="persist">
        <batch:aggregator size="100">
          <db:bulk-insert config-ref="DB">
            <db:sql>INSERT INTO enriched_customers
              (customerId, name, lat, lon)
              VALUES (:customerId, :name, :lat, :lon)</db:sql>
          </db:bulk-insert>
        </batch:aggregator>
      </batch:step>

    </batch:process-records>

    <batch:on-complete>
      <logger level="INFO"
        message="#['Done: ' ++ payload.successfulRecords ++ '/' ++ payload.totalRecords]"/>
    </batch:on-complete>
  </batch:job>
</flow>

<batch:job jobName="customerEtl"
           blockSize="200"            <!-- records per block; default 100 -->
           maxConcurrency="8"          <!-- cap parallel blocks; default ≈ 2×cores -->
           maxFailedRecords="-1">      <!-- -1 = never stop; 0 = stop on first failure -->
  <batch:process-records> <!-- ... --> </batch:process-records>
</batch:job>

<db:select config-ref="SqlServer" fetchSize="5000">   <!-- cursor fetch, not all-at-once -->
  <db:sql>SELECT * FROM customers
    WHERE lastModified >= :lastRun ORDER BY customerId</db:sql>
  <db:input-parameters>#[{ lastRun: vars.lastRunTimestamp }]</db:input-parameters>
</db:select>

%dw 2.0
output application/json
var emailRe = /^[\w.+-]+@[\w-]+\.[\w.-]+$/
// inside a batch step, payload is ONE record
---
{
  customerId: payload.customerId,
  firstName:  (payload.fullName splitBy " ")[0],
  lastName:   (payload.fullName splitBy " ")[-1],
  email:      payload.email,
  status:     payload.statusCode match {
    case "01" -> "ACTIVE"
    case "02" -> "INACTIVE"
    else      -> "UNKNOWN"
  },
  source:       "LEGACY",
  migratedDate: now()
}

<batch:job jobName="customerMigration" blockSize="500"
           maxConcurrency="8" maxFailedRecords="-1">
  <batch:process-records>

    <!-- step 1: transform; raise-error makes a bad record "failed" -->
    <batch:step name="validateTransform">
      <ee:transform> <!-- the DataWeave above --> </ee:transform>
      <validation:is-true
        expression="#[payload.email matches /^[\w.+-]+@[\w-]+\.[\w.-]+$/ and payload.firstName != null]"
        message="Invalid email or missing name"/>
    </batch:step>

    <!-- step 2: good records → Snowflake, in bulk -->
    <batch:step name="loadWarehouse" acceptPolicy="NO_FAILURES">
      <batch:aggregator size="500">
        <db:bulk-insert config-ref="Snowflake">
          <db:sql>INSERT INTO customers
            (customerId, firstName, lastName, email, status)
            VALUES (:customerId, :firstName, :lastName, :email, :status)</db:sql>
        </db:bulk-insert>
      </batch:aggregator>
    </batch:step>

    <!-- step 3: rejects → dead-letter table -->
    <batch:step name="deadLetter" acceptPolicy="ONLY_FAILURES">
      <db:insert config-ref="SqlServer">
        <db:sql>INSERT INTO etl_dead_letter
          (customerId, payload, errorReason, ts)
          VALUES (:id, :data, :reason, :ts)</db:sql>
        <db:input-parameters>#[{
          id:     payload.customerId,
          data:   write(payload, 'application/json'),
          reason: error.description default 'validation failed',
          ts:     now()
        }]</db:input-parameters>
      </db:insert>
    </batch:step>

  </batch:process-records>

  <!-- runs once: advance the watermark for next night's incremental run -->
  <batch:on-complete>
    <db:update config-ref="SqlServer">
      <db:sql>UPDATE etl_control SET lastRun = :ts WHERE job = 'customer'</db:sql>
      <db:input-parameters>#[{ ts: now() }]</db:input-parameters>
    </db:update>
    <logger level="INFO"
      message="#['Loaded ' ++ payload.successfulRecords ++ ', failed ' ++ payload.failedRecords]"/>
  </batch:on-complete>
</batch:job>