Spring Integration Filters (Validation Filter) :Lab-8

-

SI also comes with a ready-made validating filter. When dealing with XML messages that
are supposed to conform to an XML schema, it is often helpful to weed out XML messages
that do not conform to the schema. Using SI’s built-in validating filter, all you have to do is
specify the location of the appropriate schema (.xsd) file and XML messages are filtered out
of a channel when they do not align with the schema definition.

Add a validating filter. By this point in the tutorials, you have started to work
enough with SI applications and SI components to start to get a feel for how they
are configured and how the framework works. So it is time to put some of those
skills to the test. Replace the XPath filter with a validating filter to accept only
those XML shiporder messages that comply with the schema you saw in step 8.1.
The basic template is shown below. You need to supply the in and out message
channels along with the location of the schema file it is to use to do the filtering.

spring-integration-xml-xpath-filter.xml

<?xml version="1.0" encoding="UTF-8"?><beans xmlns="http://www.springframework.org/schema/beans"	xmlns:int="http://www.springframework.org/schema/integration"	xmlns:int-file="http://www.springframework.org/schema/integration/file"	xmlns:int-mail="http://www.springframework.org/schema/integration/mail"	xmlns:int-xml="http://www.springframework.org/schema/integration/xml"	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int-stream="http://www.springframework.org/schema/integration/stream"	xsi:schemaLocation="	http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd	http://www.springframework.org/schema/integration http://www.springframework.org/schema/integration/spring-integration.xsd	http://www.springframework.org/schema/integration/stream http://www.springframework.org/schema/integration/stream/spring-integration-stream.xsd	http://www.springframework.org/schema/integration/file http://www.springframework.org/schema/integration/file/spring-integration-file.xsd	http://www.springframework.org/schema/integration/xml http://www.springframework.org/schema/integration/xml/spring-integration-xml.xsd">	<!-- Adapter for reading files -->	<int-file:inbound-channel-adapter id="producer-file-adapter"		channel="inboundChannel" directory="file:c://inboundXML"		prevent-duplicates="true">		<int:poller fixed-rate="5000" />	</int-file:inbound-channel-adapter>	<int:channel id="inboundChannel" />	<int-file:file-to-string-transformer		id="file-2-string-transformer" input-channel="inboundChannel"		output-channel="xml-inboundChannel" charset="UTF-8" />	<int:channel id="xml-inboundChannel" />		<int-xml:validating-filter id="validation-filter"		input-channel="xml-inboundChannel" output-channel="outboundChannel"		schema-location="META-INF/shiporder.xsd">	</int-xml:validating-filter>	<int:channel id="outboundChannel" />	<!-- Adapter for writing files -->	<int-file:outbound-channel-adapter		channel="outboundChannel" id="consumer-file-adapter" directory="file:c://outbound" />	<int:poller id="defaultPoller" default="true"		max-messages-per-poll="5" fixed-rate="200" /></beans>

XmlXPathFilterStartup.java

package com.dpq.filter.containers;import org.springframework.context.support.ClassPathXmlApplicationContext;import org.springframework.integration.support.MessageBuilder;import org.springframework.messaging.Message;import org.springframework.messaging.MessageChannel;public class XmlXPathFilterStartup {	public static void main(String[] args) {		ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(				"spring-integration-xml-xpath-filter.xml");		System.out.println("Contanier started.....");		while(true) {					}			}}

shiporder.xsd

<?xml version="1.0" encoding="UTF-8" ?><xs:schema xmlns:xs="http://www.w3.org/2001/XMLSchema"><xs:element name="shiporder">  <xs:complexType>    <xs:sequence>      <xs:element name="orderperson" type="xs:string"/>      <xs:element name="shipto">        <xs:complexType>          <xs:sequence>            <xs:element name="name" type="xs:string"/>            <xs:element name="address" type="xs:string"/>            <xs:element name="city" type="xs:string"/>            <xs:element name="country" type="xs:string"/>          </xs:sequence>        </xs:complexType>      </xs:element>      <xs:element name="item" maxOccurs="unbounded">        <xs:complexType>          <xs:sequence>            <xs:element name="title" type="xs:string"/>            <xs:element name="note" type="xs:string" minOccurs="0"/>            <xs:element name="quantity" type="xs:positiveInteger"/>            <xs:element name="price" type="xs:decimal"/>          </xs:sequence>        </xs:complexType>      </xs:element>    </xs:sequence>    <xs:attribute name="orderid" type="xs:string" use="required"/>  </xs:complexType></xs:element></xs:schema>

Below are few XMLs which will be pushed into ‘/Users/dpq/Desktop/techWorrk/fileInbound‘

shiporder1.xml

<?xml version="1.0" encoding="UTF-8"?><shiporder orderid="889923"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:noNamespaceSchemaLocation="shiporder.xsd">  <orderperson>John Smith</orderperson>  <shipto>    <name>George Blatt</name>    <address>Elm St</address>    <city>New Orleans</city>    <country>USA</country>  </shipto>  <item>    <title>Hide your heart</title>    <quantity>2</quantity>    <price>19.80</price>  </item></shiporder>

shiporder2.xml

<?xml version="1.0" encoding="UTF-8"?><shiporder orderid="889923"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:noNamespaceSchemaLocation="shiporder.xsd">  <orderperson>John Smith</orderperson>  <shipto>    <name>Ola Nordmann</name>    <address>Langgt 23</address>    <city>4000 Stavanger</city>    <country>Norway</country>  </shipto>  <item>    <title>Empire Burlesque</title>    <note>Special Edition</note>    <quantity>1</quantity>    <price>10.90</price>  </item>  <item>    <title>Hide your heart</title>    <quantity>1</quantity>    <price>9.90</price>  </item></shiporder>

shiporder3.xml

<?xml version="1.0" encoding="UTF-8"?><shiporder orderid="889923"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:noNamespaceSchemaLocation="shiporder.xsd">   <shipto>    <name>Ola Nordmann</name>    <address>Langgt 23</address>    <city>4000 Stavanger</city>    <country>Norway</country>  </shipto>  <item>    <title>Empire Burlesque</title>    <note>Special Edition</note>    <quantity>1</quantity>    <price>10.90</price>  </item>  <item>    <title>Hide your heart</title>    <quantity>1</quantity>    <price>9.90</price>  </item></shiporder>

If we push all above XMLs, See that the accepted messages are now in the folder. Rejected messages (files with names that do not contain a < with content of USA) have been filtered and processed

3rd XML will not be processed because <orderperson> tag is missing but same is part of xsd hence it has to be validated by xsd so this message will be excluded

Popular posts from this blog

Window function in PySpark with Joins example using 2 Dataframes (inner join)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Solution: from pyspark.sql.functions import col # this can be done without using window function creditCardData = spark.read.json(“card_transactions.json”) useridMaxSpendDF = df.groupby(‘user_id’).max(‘amount’) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“max(amount)”,”max_amount”) useridMaxSpendDF=useridMaxSpendDF.withColumnRenamed(“user_id”,”m_user_id”) cond = [creditCardData.user_id == useridMaxSpen...
Read more

Complex SQL: fetch the users who logged in consecutively 3 or more times (lead perfect example)

-
From the login_details table, fetch the users who logged in consecutively 3 or more times. Table Name : LOGIN_DETAILS Approach : We need to fetch users who have appeared 3 or more times consecutively in login details table. There is a window function which can be used to fetch data from the following record. Use that window function to compare the user name in current row with user name in the next row and in the row following the next row. If it matches then fetch those records. –Table Structure: drop table login_details; create table login_details( login_id int primary key, user_name varchar(50) not null, login_date date); delete from login_details; insert into login_details values (101, ‘Michael’, current_date), (102, ‘James’, current_date), (103, ‘Stewart’, current_date+1), (104, ‘Stewart’, current_date+1), (105, ‘Stewart’, current_date+1), (106, ‘Michael’, current_date+2), (107, ‘Michael’, cur...
Read more

Credit Card Data Analysis using PySpark (how to use auto broadcast join after disabling it)

-
Input details: #● File has json records #● Each record has fields: #○ user_id #○ card_num #○ merchant #○ category #○ amount #○ ts ### Below analysis to be done Sample data: +------+--------+---------+--------+----------+-------+|amount|card_num| category|merchant| ts|user_id|+------+--------+---------+--------+----------+-------+| 243| C_108| food| M_102|1579532902| U_104|| 699| C_106|cosmetics| M_103|1581759040| U_103|| 228| C_104| children| M_110|1584161986| U_103| Application: Here we will disable auto broadcast and then join 2 data frames and will see execution plan which should use broadcast join Solution: from pyspark.sql import SparkSession from pyspark.sql.functions import broadcast spark = SparkSession.builder.master(‘local[2]’)\ .appName(‘RDD_Methods_Examples’)\ .getOrCreate() print(spark.conf.get(“spark.sql.autoBroadcastJoinThreshold”)) spark.conf.set(“spark.sql.autoBroadcastJoinThreshold”, -1) pr...
Read more