MuleSoft Integration Platform Basics: “Optimizing Workflows: From File Reading to JSON Conversion and Order Transaction Checks” — Part 4

8 min readApr 22, 2024

This article has been prepared by our Salesforce Engineer, Atharva Toke, member of the Thinqloud Engineering team.

In this segment of our MuleSoft basics series, we will explore the process of reading CSV and Excel files, inserting records into a database, and implementing duplication checks. These essential steps in data integration with MuleSoft are crucial for effectively managing and utilizing data within your organization.

A Mule Flow to read transactions from a CSV file and add transactions into a PostgreSQL Database order table.

Create a Mule Project: -

Create a Mule Project: In Anypoint Studio, go to File > New > Mule Project.
Enter a name for your project and click Finish.

Configure a Source:-

In the Mule Palette view, select “Scheduler” as the source.
Drag it onto the Studio canvas.
Specify the frequency as per the requirement. (For Above flow we had taken 2 minutes means every 2 minutes flow will execute)
- Fixed Frequency ⇒ Selected
- Cron

Read the CSV File: -

In the Mule Palette view, select “FTP Connector-Read” as the process.
Drag it onto the Studio canvas.
Using the “FTP Connector-Read” element on the process block to read files from the FTP(Filezilla) server.

Transform CSV to JSON:-

In the Mule Palette view, select “Transform Message (Core)” as the process.
Drag it onto the Studio canvas.
“Transform Message (Core)” to transform the JSON format from CSV file.

Using Script: -

%dw 2.0
output application/json
---
payload map ( payload01 , indexOfPayload01 ) -> {
 purchase_order_number: payload01.purchase_order_number,
 item: payload01.item
}

Add Logger Component to Your Flow & Configure the Logger: -

The Logger component helps monitor and debug your Mule application by logging important information like error messages, status notifications, payloads, etc.
Specify the message you want to log by setting the “message” attribute in the Logger configuration. e.g., here we are giving passing message attribute with #[pauload] to log the data so that we can see data or errors can be shown in the console.

Add the Database Connector to Your Project: -

In the Mule Palette view, click on Search in Exchange.
Search for “Database Connector” and select it.
Drag and drop the “Bulk Insert” connector as we want to insert records to the PostgreSQL in bulk-single transaction rather than multiple single instances.

Add PostgreSQL Connector Dependency:-

In your Mule Canvas, navigate to the database connector and configuration with the PostgreSQL Database, by clicking on the {+} icon.
Add the PostgreSQL Connector (Maven) dependency by adding JDBC Driver:


<!-- https://mvnrepository.com/artifact/org.postgresql/postgresql -->
<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.7.2</version>
</dependency>

Configure Database Connection:-

Enter the necessary connection details like:-
- URL ⇒ jdbc:postgresql://178.18.16.208:5432/mulesoft_db
- Driver Class Name ⇒ org.postgresql.Driver
- User ⇒ As per Requirement
- Passwords ⇒ As per Requirement

Test the Connection: -

Run your application in Anypoint Studio to test the MySQL connection.
Verify that you can connect to the MySQL database and perform operations successfully.

Close the Connection:-

Ensure that you close the MySQL connection after executing your queries to release resources properly.
Add Logger Component again to your Flow & Configure the Logger: -
The Logger component helps monitor and debug your Mule application by logging important information like error messages, status notifications, payloads, etc.
Specify the message you want to log by setting the “message” attribute in the Logger configuration. E.g. here we are giving passing message attribute with #[pauload] to log the data so that we can see data or errors can be shown in the console.

Additional Logger adding to check the time required for the process of the flow.

Result:-

Code for the configuration- pom.xml (please adjust some part from this snippet based on your needs)


<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:db="http://www.mulesoft.org/schema/mule/db" xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core"
 xmlns:ftp="http://www.mulesoft.org/schema/mule/ftp"
 xmlns="http://www.mulesoft.org/schema/mule/core" xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/ftp http://www.mulesoft.org/schema/mule/ftp/current/mule-ftp.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd
http://www.mulesoft.org/schema/mule/db http://www.mulesoft.org/schema/mule/db/current/mule-db.xsd">
 <ftp:config name="FTP_Config" doc:name="FTP Config" doc:id="f4f002a9-10cd-4221-97e9-991728a421d0" >
  <ftp:connection workingDir="ftpdatabasejson" host="182.16.18.203" username="FTP-Server" password="$ecure#890" />
 </ftp:config>
 <db:config name="Database_Config" doc:name="Database Config" doc:id="e234c8f8-eb2f-42b0-81b0-e4cacf774ceb" >
  <db:generic-connection url="jdbc:postgresql://182.16.18.203:5432/mulesoft_db" driverClassName="org.postgresql.Driver" user="postgres" password="postgres" />
 </db:config>
 <db:config name="Database_Config1" doc:name="Database Config" doc:id="8a8fd054-ba88-4f19-a96b-7625bbfeefd9" >
  <db:my-sql-connection host="182.16.18.203" port="3306" user="root" password="M1cr0s0ft@123" database="mysql" />
 </db:config>
 <flow name="ftp-databaseFlow" doc:id="1fa3fa48-0299-415f-96d4-6ffa94e035c8" >
  <scheduler doc:name="Scheduler" doc:id="5b563314-34a4-4d24-b088-a65e161ed8fe" >
   <scheduling-strategy >
    <fixed-frequency frequency="5" timeUnit="MINUTES"/>
   </scheduling-strategy>
  </scheduler>
  <ftp:read doc:name="Read csv" doc:id="70a27614-a9e4-4225-8e25-7c8646f0ba6c" config-ref="FTP_Config" path="ftpcsv - bulk.csv" outputMimeType="application/csv"/>
  <ee:transform doc:name="Transform Message" doc:id="89a60cae-bde4-4f72-9988-cb177dec15a0" >
   <ee:message >
    <ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload map ( payload01 , indexOfPayload01 ) -> {
 purchase_order_number: payload01.purchase_order_number,
 item: payload01.item
}]]></ee:set-payload>
   </ee:message>
  </ee:transform>
  <logger level="INFO" doc:name="Logger" doc:id="d2753b3a-6968-41a3-9725-52cff1bc0cfd" message="#[payload]"/>
  <db:bulk-insert doc:name="Bulk insert on ftp_database_bulk" doc:id="5d455105-6472-4d05-b23e-b79c451eb6f5" config-ref="Database_Config">
   <db:sql ><![CDATA[INSERT INTO public.poc_ftp_database_bulk(
 purchase_order_number, item)
 VALUES (:purchase_order_number, :item);]]></db:sql>
  </db:bulk-insert>
  <logger level="INFO" doc:name="Logger" doc:id="b588440f-7e84-46d1-8a92-bba386967516" message="#[payload]"/>
  <logger level="INFO" doc:name="End Flow" doc:id="792ff999-c174-41fe-8d21-3c8f9b363e64" message="End flow #[flow.name]. Duration: #[server.dateTime.getTime() - vars.startTime]" />
 </flow>
</mule>

By following these steps, you can create a seamless connection between your Mule application in Anypoint Studio and Local System and read the CSV file from the folder and transform that CSV to JSON and then Bulk inserting those records into the Database Table.

2. A Mule Flow to check duplication on the order transactions which are being read from the network folder.

Create a Mule Project: -

Create a Mule Project: In Anypoint Studio, go to File > New > Mule Project.
Enter a name for your project and click Finish.

Configure a Source: -

In the Mule Palette view, select “Scheduler” as the source.
Drag it onto the Studio canvas.
Specify the frequency as per the requirement. (For Above flow we had taken 2 minutes means every 2 minutes flow will execute)
- Fixed Frequency ⇒ Selected
- Cron

Read the CSV File: -

In the Mule Palette view, select “FTP Connector-Read” as the process.
Drag it onto the Studio canvas.
Using the “FTP Connector-Read” element on the process block to read files from the FTP(Filezilla) server.

Transform CSV to JSON: -

In the Mule Palette view, select “Transform Message (Core)” as the process.
Drag it onto the Studio canvas.
“Transform Message (Core)” to transform the JSON format from CSV file.

Using Script:-

payload map ( payload01 , indexOfPayload01 ) -> {
 id: payload01.id,
 name: payload01.name,
 city: payload01.city
}

Add Logger Component to Your Flow & Configure the Logger: -

The Logger component helps monitor and debug your Mule application by logging important information like error messages, status notifications, payloads, etc.
Specify the message you want to log by setting the “message” attribute in the Logger configuration. e.g., here we are giving passing message attribute with #[pauload] to log the data so that we can see data or errors can be shown in the console.

Use DataWave function of Transform Element:-
“Transform Message (Core)” to find out the duplicates in the records as per id.


%dw 2.0
output application/java
---
distinctBy(payload, (val, valIndex) -> val.id)

Again, using Transform Element to change the sorted record into JSON: -
“Transform Message (Core)” to transform the CSV format from JSON format.

Using Script:-

%dw 2.0
output application/xlsx
---
payload map ( payload01 , indexOfPayload01 ) -> {
 id: payload01.id,
 name: payload01.name,
 city: payload01.city
}

Add Local Connector to write Excel file:-

In the Mule Palette view, select ““Write”” as the source.
Drag it onto the Studio canvas.
Specify the path where you want to write the Excel file.

Add Logger to Analyze the data.

Result:-

In the image on the left, you can see an Excel Sheet that contains duplicate data with matching IDs. However, after running the Mule Flow, the new Excel Sheet on the right is free from any duplicates, with the condition that the IDs must not be the same.

Code for the configuration- pom.xml (please adjust some part from this snippet based on your needs)

<?xml version="1.0" encoding="UTF-8"?>
<mule xmlns:ee="http://www.mulesoft.org/schema/mule/ee/core" xmlns:file="http://www.mulesoft.org/schema/mule/file"
 xmlns="http://www.mulesoft.org/schema/mule/core"
 xmlns:doc="http://www.mulesoft.org/schema/mule/documentation" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://www.mulesoft.org/schema/mule/core http://www.mulesoft.org/schema/mule/core/current/mule.xsd
http://www.mulesoft.org/schema/mule/file http://www.mulesoft.org/schema/mule/file/current/mule-file.xsd
http://www.mulesoft.org/schema/mule/ee/core http://www.mulesoft.org/schema/mule/ee/core/current/mule-ee.xsd">
 <file:config name="File_Config" doc:name="File Config" doc:id="6c4b9de9-c28d-48f5-9e3e-8ba0258dfd80" >
  <file:connection workingDir="E:\Backup_Office_server\FTP\ftpsql" />
 </file:config>
 <file:config name="File_Config1" doc:name="File Config" doc:id="1c24b2ee-b0db-479c-9f55-6289ce14a2bd" >
  <file:connection workingDir="E:\Backup_Office_server\Projects_Files\poc-duplication-record-check" />
 </file:config>
 <file:config name="File_Config2" doc:name="File Config" doc:id="9e1dcccc-03ea-4d1e-afd1-4cdf54188b08" >
  <file:connection workingDir="E:\Backup_Office_server\Projects_Files\poc-duplication-record-check" />
 </file:config>
 <flow name="poc-duplication-records-checkFlow" doc:id="84bdc215-9129-4c59-973e-224593e8befa" >
  <scheduler doc:name="Scheduler" doc:id="0a0ecc04-a977-4bb9-9823-ae9cd62e9190" >
   <scheduling-strategy >
    <fixed-frequency frequency="3" timeUnit="MINUTES"/>
   </scheduling-strategy>
  </scheduler>
  <file:read doc:name="Read" doc:id="2c3c1017-26b8-4a27-bb02-8571fd319af9" config-ref="File_Config1" path="E:\Backup_Office_server\Projects_Files\poc-duplication-record-check\dummyrecords.csv"/>
  <logger level="INFO" doc:name="Logger" doc:id="289143d9-307e-40f5-81b2-7b046ff35a83" message="#[payload]"/>
  <ee:transform doc:name="Transform Message" doc:id="608b28e1-5b32-4271-bcbd-9eb14e1569f5" >
   <ee:message >
    <ee:set-payload ><![CDATA[%dw 2.0
output application/json
---
payload map ( payload01 , indexOfPayload01 ) -> {
 id: payload01.id,
 name: payload01.name,
 city: payload01.city
}]]></ee:set-payload>
   </ee:message>
  </ee:transform>
  <logger level="INFO" doc:name="Logger" doc:id="74b4131e-3a1f-409e-a4a4-8b2684b1feff" message="#[payload]"/>
  <ee:transform doc:name="Transform Message" doc:id="a0b0493b-3a69-4114-861a-0871cb2ee045" >
   <ee:message >
    <ee:set-payload ><![CDATA[%dw 2.0
output application/java
---
distinctBy(payload, (val, valIndex) -> val.id)]]></ee:set-payload>
   </ee:message>
  </ee:transform>
  <ee:transform doc:name="Transform Message" doc:id="3145219e-bf27-491a-b345-528daa0d2d4f" >
   <ee:message >
    <ee:set-payload ><![CDATA[%dw 2.0
output application/csv
---
payload map ( payload01 , indexOfPayload01 ) -> {
 id: payload01.id,
 name: payload01.name,
 city: payload01.city
}]]></ee:set-payload>
   </ee:message>
  </ee:transform>
  <file:write doc:name="Write" doc:id="e10bb9b8-bdcd-49cb-8c1c-4e288d6811eb" config-ref="File_Config2" path="E:\Backup_Office_server\Projects_Files\poc-duplication-record-check\duplicate_record_check.csv"/>
  <logger level="INFO" doc:name="Logger" doc:id="a3bec476-2dbe-4449-a3d4-f754cb24999a" />

By following these steps, you can effortlessly establish a seamless connection between your Mule application in Anypoint Studio and the Network folder on the Internal Server. This connection allows you to read CSV or any other files and then convert those records into JSON format. Additionally, you can perform a duplication check based on specific parameters and write the records on your Local system.

Coming Up Next….

This continuation blog post marks the continuing of an exciting journey, introducing Part 4 of our MuleSoft base knowledge series and offering a tantalizing glimpse into the captivating content awaiting readers in future installments. Stay tuned for a wealth of engaging topics and insightful discussions on MuleSoft in the upcoming blogs!

MuleSoft Integration Platform Basics: “Enhancing Monitoring and Logging: Error Handling, Log Saving” — Part 5
MuleSoft Integration Platform Basics: “Empowering Connectivity: API Gateway Setup and Database Invoicing with Fast API Integration and Real-Time Project Monitoring on CloudHub” — Part 6
Access Part I of the series “A Catalyst for Seamless Integration and Business Transformation” — here
Access Part 2 of the series “Streamlining Data Integration: A Guide to Setting up Connectors and Network Sharing” — here
MuleSoft Integration Platform Basics: “Efficient Data Processing — here

Follow us & Connect at:-

LinkedIn:- https://in.linkedin.com/company/thinqloud

medium.com:- Thinqloud Solutions — Medium

Company Site:- https://www.thinqloud.com/

Please let us know if you have any suggestions, comments or just general feedback for us!

MuleSoft Integration Platform Basics: “Optimizing Workflows: From File Reading to JSON Conversion and Order Transaction Checks” — Part 4

Written by Thinqloud Solutions