DEV Community: DarrylBryson

XML Batch Data Processing

DarrylBryson — Wed, 31 Mar 2021 23:27:55 +0000

XML Batch Data Processing

For decades xml was the dominate format many applications used to communicate with one another. In recent years xml has become less and less popular as a data format for semi-structured data cases. Now in modern cloud native, microservice applications json or yaml has become the norm. However this doesn't change the fact that much of the systems, tooling, and business process are still in place. XML is also the defacto 'neutral' format for converting industry specific file types. As a professional analyst being able to process xml as easily as csv is a must.

In this post we'll create a batch data processing script for a typical lab reporting use case. The scenario, a technician, Susian, at the end of each day analyzes the departments test results. The lab equipment outputs test results as xml. Susian requires the xml to be converted into csv in order to feed an Excel report she was given.

I'll be using a sample lab-measurement-dataset I auto generated in a previous post here.

Contents

XML Batch Data Processing
- Convert XML to CSV
- Parsing XML: Root element
- Parsing XML: First level elements
- Parsing XML: Sub-elements
- XML Parser Class
- Batch Execution
- Save Converted Dataset
- TL;DR

Convert XML to CSV

Example XML File

<?xml version="1.0" ?>
<DataFiles>
  <DataFile id="d7ecf46df41f4355a23c42c4607266c7">
    <machine_id>Machine_10</machine_id>
    <test_id>3c0c95f773bf426585a3a68642d2d41a</test_id>
    <technician>Brett Kerr</technician>
    <test_routine>c/f/a/b</test_routine>
    <batched>Yes</batched>
    <loc_1>
      <x_offset>8.63</x_offset>
      <y_offset>0.39661</y_offset>
      <z_offset>3736</z_offset>
    </loc_1>
    <loc_2>
      <x_offset>-4.94</x_offset>
      <y_offset>0.964363</y_offset>
      <z_offset>3182</z_offset>
    </loc_2>
  </DataFile>
</DataFiles>

Target Table Structure

machine_id	test_id	technician	test_routine	batched	measurement_location_id	x_offset	y_offset	z_offset
Machine_03	87729203dfce4e9da7efbbb985c83bd9	Stacey Simpson	d/a/f	Yes	loc_1	47.81	1.878128	871
Machine_03	87729203dfce4e9da7efbbb985c83bd9	Stacey Simpson	d/a/f	Yes	loc_2	-2.1	0.995672	4694
Machine_03	98910cae86864c3697e6e7fccfb8cc33	Brittney Gray	f/d/a	Yes	loc_1	2.5	0.236024	3034

Parsing XML: Root element

To handel the xml parsing we'll use the import xml.etree.ElementTree package. The key to parsing xml is creating a class representing a single file and methods that translate each element. In the example file the first element to isolate will be each DataFile for the DataFiles collection:

import xml.etree.ElementTree as ET
import uuid
import pandas as pd
import glob

file_path = '/sample-data-set/auto-gen/xml/a8971cf83bd84fd1b366bfb312278021.xml'
with open(file_path) as f:
    tree = ET.parse(f)
    rootElem = tree.getroot()

    # Select each DataFile
    for data_file in rootElem.findall('DataFile'):
        print(data_file.get('id'))

>>> output:
d7ecf46df41f4355a23c42c4607266c7

Parsing XML: First level elements

Next extract the 'Header' information for each test

file_path = '/sample-data-set/auto-gen/xml/a8971cf83bd84fd1b366bfb312278021.xml'
with open(file_path) as f:
    tree = ET.parse(f)
    rootElem = tree.getroot()
for data_file in rootElem.findall('DataFile'):
    m_id = data_file.find('machine_id').text
    test_id = data_file.find('test_id').text
    tech_id = data_file.find('technician').text
    test_routine = data_file.find('test_routine').text
    batched = data_file.find('batched').text

    header_ar = [m_id, test_id, tech_id, test_routine, batched]
    print(header_ar)

>>> output:
['Machine_03', '9e0f95807ed44a468271eb6d3ff85a44', 'Jennifer Johnson', 'd', 'N/A']

Parsing XML: Sub-elements

Finally, extract the measurement data for loc_1 and loc_2

file_path = '/sample-data-set/auto-gen/xml/a8971cf83bd84fd1b366bfb312278021.xml'

def parse_measurement_location(dataFile, loc_name):
    res = []
    measurement_loc = dataFile.find(loc_name)
    measurement_loc_id = measurement_loc.tag
    x_m = measurement_loc.find('x_offset').text
    y_m = measurement_loc.find('y_offset').text
    z_m = measurement_loc.find('z_offset').text

    return [measurement_loc_id,x_m,y_m,z_m]

with open(file_path) as f:
    tree = ET.parse(f)
    rootElem = tree.getroot()
for data_file in rootElem.findall('DataFile'):
    file_ar = []
    m_id = data_file.find('machine_id').text
    test_id = data_file.find('test_id').text
    tech_id = data_file.find('technician').text
    test_routine = data_file.find('test_routine').text
    batched = data_file.find('batched').text

    header_ar = [m_id, test_id, tech_id, test_routine, batched]

    loc_ar = parse_measurement_location(data_file, 'loc_1')
    file_ar.append(header_ar + loc_ar)

    loc_ar = parse_measurement_location(data_file, 'loc_2')
    file_ar.append(header_ar + loc_ar)
    print(file_ar)

>>> output:
[['Machine_03', '9e0f95807ed44a468271eb6d3ff85a44', 'Jennifer Johnson', 'd', 'N/A', 'loc_1', '17.35', '1.3074', '752'], 

['Machine_03', '9e0f95807ed44a468271eb6d3ff85a44', 'Jennifer Johnson', 'd', 'N/A', 'loc_2', '2.68', '0.575979', '4483']]

XML Parser Class

We then place the transformation logic into a class that can be called for any file:

# Parsing class
class Xml_Parser:
    def __init__(self):
        self.ResultAr = []
        return
    def parse_datafiles(self,file_id,rootElem):
        res = []
        for data_file in rootElem.findall('DataFile'):
            file_ar = self.parse_file(data_file)
            res += file_ar
        return res
    def parse_file(self, data_file):

        file_ar = []
        m_id = data_file.find('machine_id').text
        test_id = data_file.find('test_id').text
        tech_id = data_file.find('technician').text
        test_routine = data_file.find('test_routine').text
        batched = data_file.find('batched').text

        header_ar = [m_id, test_id, tech_id, test_routine, batched]

        loc_ar = self.parse_measurement_location(data_file, 'loc_1')
        file_ar.append(header_ar + loc_ar)

        loc_ar = self.parse_measurement_location(data_file, 'loc_2')
        file_ar.append(header_ar + loc_ar)

        return file_ar
    def parse_measurement_location(self, dataFile, loc_name):
        res = []
        measurement_loc = dataFile.find(loc_name)
        measurement_loc_id = measurement_loc.tag
        x_m = measurement_loc.find('x_offset').text
        y_m = measurement_loc.find('y_offset').text
        z_m = measurement_loc.find('z_offset').text

        return [measurement_loc_id,x_m,y_m,z_m]

Batch Execution

To do a batch conversion on a set of files we'll need to:

Parse a given file by executing our parsing class
Find all test files


# Conversion executor function
def convert_xml_to_list(file_path):
    with open(file_path) as f:
        tree = ET.parse(f)
        root = tree.getroot()
        parser_obj = Xml_Parser()
        xml_list = parser_obj.parse_datafiles(f,root)
    return xml_list

# Batch executor
def batch_convert_xml_to_df(xml_dir, dataset_columns,file_limit=-1):
    i=0
    converted_dataset = []
    # Recursively convert each target file
    for filepath in glob.iglob(xml_dir, recursive=True):
        c_ds = convert_xml_to_list(filepath)
        converted_dataset+=c_ds

        i+=1
        if (i >= file_limit) and (file_limit>=0): break

    df = pd.DataFrame(converted_dataset, columns = dataset_columns) 
    return df

Save Converted Dataset

The last step is to bring all the pieces together convert and save a folder of xml files to a single csv.

measurement_file_path = '/sample-data-set/auto-gen/xml/*.xml'
measurement_columns = ['machine_id','test_id','technician','test_routine','batched','measurement_location_id','x_offset','y_offset','z_offset']
file_process_limit = 20 # Set to -1 for unlimited

converted_file_dir = '/sample-data-set/auto-gen/converted'

lab_measurement_df = batch_convert_xml_to_df(measurement_file_path, measurement_columns, file_process_limit)

# Save dataset to csv using unique name
destPath = '/'.join([converted_file_dir,str(uuid.uuid4().hex)])
destPath = '.'.join([destPath,'csv'])

lab_measurement_df.to_csv(destPath, index = False, header=True)

lab_measurement_df.head()

	machine_id	test_id	technician	test_routine	batched	measurement_location_id	x_offset	y_offset	z_offset
0	Machine_03	87729203dfce4e9da7efbbb985c83bd9	Stacey Simpson	d/a/f	Yes	loc_1	47.81	1.878128	871
1	Machine_03	87729203dfce4e9da7efbbb985c83bd9	Stacey Simpson	d/a/f	Yes	loc_2	-2.1	0.995672	4694
2	Machine_03	98910cae86864c3697e6e7fccfb8cc33	Brittney Gray	f/d/a	Yes	loc_1	2.5	0.236024	3034
3	Machine_03	98910cae86864c3697e6e7fccfb8cc33	Brittney Gray	f/d/a	Yes	loc_2	-13.38	0.795762	4337
4	Machine_03	f505ffc52ad34b56a9f7f95451a813c3	John King	c/f/e	No	loc_1	44.2	1.982816	830

TL;DR

Complete script here

Auto Generate Lab Measurement Dataset

DarrylBryson — Wed, 24 Mar 2021 08:48:13 +0000

Auto Generate Lab Measurement Dataset

view Notebook

A typical data source found in most industrial work places are lab measurements. This type of data usually has a well defined structure and is a mixture of categorical and numeric data types. Measurement ranges and file format can vary greatly between use cases. This article covers how to generate a simple dataset that can be customized to simulate a wide range of use cases.

Data Structure
Packages
- Faker
- xml.etree.ElementTree
- Pandas
Generate Dataset
Convert Dataset to XML
Convert Dataset to Pandas DataFrame
Save File
- csv
- xml
- parquet

Sample Data Structure

Field	DataType	Min	Max
machine_id	string
test_id	uuid
technician	string
test_routine	categorical
batched	categorical
loc_1	dictionary
loc_1_x_offset	decimal	-15	15
loc_1_y_offset	decimal	-1	1
loc_1_z_offset	int	2500	5000
loc_2	dictionary
loc_2_x_offset	decimal	-15	15
loc_2_y_offset	decimal	-1	1
loc_2_z_offset	int	2500	5000

Packages

Faker: Is a python package for creating, as the name implies, fake data. This library not only generates random numbers and strings but more complex elements like addresses, and names.

from faker import Faker
fake = Faker()
Faker.seed(0)

xml.etree.ElementTree: Is a base library of Python that is a simple and efficient way of querying, parsing, and creating XML data.

import xml.etree.ElementTree as ET
from xml.dom import minidom

Pandas: Is the goto library for working with table-like data.

import pandas as pd

# base imports
import uuid
import os

Generate Dataset

The structure of the dataset is represented by a dictionary and uses Faker to fill in the data values. This allows for simple modifications when moving between use cases.

def generate_measurement_record():
    base_equipment_name='Machine'
    equipment_cnt = 25
    n = fake.pyint(min_value=1, max_value=equipment_cnt)

    measurement_record = {
        'machine_id': '_'.join([base_equipment_name, f'{n:02}'])
        ,'test_id':str(uuid.uuid4().hex)
        ,'technician':fake.name()
        ,'test_routine':fake.random_sample(elements=('a', 'b', 'c', 'd', 'e', 'f'))
        ,'batched':fake.random_sample(elements=('Yes', 'No', 'N/A'), length=1)[0]
        ,'loc_1':{
            'x_offset':fake.pydecimal(left_digits=2, right_digits=2, min_value=-15, max_value=15)
            ,'y_offset':fake.pydecimal(left_digits=1, right_digits=6, min_value=-1, max_value=1)
            ,'z_offset':fake.pyint(min_value=2500, max_value=5000)
        }
        ,'loc_2':{
            'x_offset':fake.pydecimal(left_digits=2, right_digits=2, min_value=-15, max_value=15)
            ,'y_offset':fake.pydecimal(left_digits=1, right_digits=6, min_value=-1, max_value=1)
            ,'z_offset':fake.pyint(min_value=2500, max_value=5000)
        }
    }
    return measurement_record

Repeatedly calling this function generates new records with random data that can be collected in a list up to a desired size.

records_cnt = 5

measurement_record = None
measurement_list=[]

for _ in range(records_cnt):
    measurement_record = generate_measurement_record()
    measurement_list.append(measurement_record) # List of nested dictionaries

Convert Dataset to XML

Next we'll need a function that will handle converting the records stored as dictionaries to xml. Then apply the function to each record in the list.

import xml.etree.ElementTree as ET

def dict_to_xml(d, r=None):
    file_id = str(uuid.uuid4().hex)
    if r is None:
        r = ET.Element('DataFile')
        r.set('id', file_id)
    if isinstance(d, dict):
        for k, v in d.items():
            s = ET.SubElement(r, k)
            dict_to_xml(v, s)
    elif isinstance(d, tuple) or isinstance(d, list):
        val = '/'.join(str(v) for v in d)
        r.text = val
    elif isinstance(d, str):
        r.text = d
    else:
        r.text = str(d)
    return r

# Convert each record
xml_dataset = ET.Element('DataFiles')
for rcd in measurement_list:
    file_id = str(uuid.uuid4().hex)
    child = ET.SubElement(xml_dataset,'DataFile')
    child.set('id', file_id)

    # Convert to xml
    element= dict_to_xml(rcd,child)

The resulting xml file can be viewed with:

from xml.dom import minidom

xml= ET.tostring(xml_dataset, encoding='unicode', method='xml')
xml= minidom.parseString(xml)
xml= xml.toprettyxml(indent='  ')

print(xml)

<?xml version="1.0" ?>
<DataFiles>
  <DataFile id="33c768494e484df9b954c4c7bae2ce68">
    <machine_id>Machine_10</machine_id>
    <test_id>3c0c95f773bf426585a3a68642d2d41a</test_id>
    <technician>Brett Kerr</technician>
    <test_routine>c/f/a/b</test_routine>
    <batched>Yes</batched>
    <loc_1>
      <x_offset>8.63</x_offset>
      <y_offset>0.39661</y_offset>
      <z_offset>3736</z_offset>
    </loc_1>
    <loc_2>
      <x_offset>-4.94</x_offset>
      <y_offset>0.964363</y_offset>
      <z_offset>3182</z_offset>
    </loc_2>
  </DataFile>
</DataFiles>

Convert Dataset to Pandas DataFrame

Pandas can directly create a DataFrame from a dictionary if the dictionary is not nested. So, the first step of conversion is to create a function to flatten the measurement dictionary structure:

import collections.abc

def flatten(d, parent_key='', sep='_'):
    items = []
    for k, v in d.items():
        new_key = parent_key + sep + k if parent_key else k
        if isinstance(v, collections.MutableMapping):
            items.extend(flatten(v, new_key, sep=sep).items())
        else:
            items.append((new_key, v))
    return dict(items)

Next, create a new list of the flattened records:

measurement_flat_list= []

for rcd in measurement_list:
    measurement_flat_list.append(flatten(rcd)) # List of single depth dictionaries

Finally, convert the list to a dataframe:

measurement_df = pd.DataFrame(measurement_flat_list)
measurement_df.head()

	machine_id	test_id	technician	test_routine	batched	loc_1_x_offset	loc_1_y_offset	loc_1_z_offset	loc_2_x_offset	loc_2_y_offset	loc_2_z_offset
0	Machine_02	ffbb9f35d5484deebab0598926194203	Stephanie Leblanc	[f, d, c, e, a]	Yes	14.86	0.607854	3627	0.63	0.86374	3828
1	Machine_20	fb824a88e0f145169b411a2180c5671c	Kevin Rogers	[b, a]	N/A	-6.14	0.390133	3198	-4.54	0.105494	3099
2	Machine_23	25d73f925ba14df5a09a11ef8fb3a68d	Robert Walters	[e, a, f, d, c]	N/A	12.73	0.410212	2874	-3.14	0.634917	2588
3	Machine_07	a3fb30a0fe8f4000aa72eb182230d236	Cathy Martinez	[f, a]	N/A	-14.69	0.650746	2915	12.33	0.231556	2794
4	Machine_21	02a27172e035464fbdf680c476312b03	Jeffrey Brown	[a, e, f, c]	No	-8.33	0.94833	4426	12.72	0.731588	3333

Save Dataset Files

XML
CSV
Parquet

# xml
destPath = 'sample_lab_measurement.xml'
# Convert to string
xml_str= ET.tostring(xml_dataset, encoding='unicode', method='xml')

# Pretty print string 
xml_str= minidom.parseString(xml_str)
xml_str= xml_str.toprettyxml(indent='  ')

# Write file
with open(destPath, 'w') as f:  # Write in file as utf-8
    f.write(xml_str)

# csv
destPath = 'sample_lab_measurement.csv'
measurement_df.to_csv(destPath, index = False, header=True)

# parquet
destPath = 'sample_lab_measurement.parquet'
measurement_df.to_parquet(destPath, index = False)

Deployment Walkthrough: Microservice Communication Interface

DarrylBryson — Mon, 15 Mar 2021 21:45:46 +0000

Deployment Walkthrough: Microservice Communication Interface

The purpose of this article is to walkthrough the setting up of infrastructure for containerized applications to communicate and access files. Containerizing an application gives great flexibility in how, where, and at what scale it’s deployed. A drawback to this versatility is the problem of discoverability. Microservice applications often rely on accessing remote files and the ability to communicate with other apps. How does an app find these resources when being deployed into different data centers and dynamically assign IP addresses?

Hard Coded Solution: Configuration files

One way to resolve this, is to collect deployment environment differences into a configuration file. A new config. file is needed per deployment environment and app. This solution is straight forward but has a significant maintenance cost as the number of apps increases or are reused in multiple locations (1 Config per Env per App). This can really add up when an enterprise solution can rely on several dozen microservices.

Diagram 1-1: Direct hard coded addressing

Generalized Solution: Communication Interfaces

A solution for generalized deployment scenarios is to deploy containerizes that can interface between the microservices and the environments they're deployed into. Interfaces deployed as containers are instantly discoverable to other apps in the same Docker “network”.

The general concept is to utilize an sftp server and a messaging queue to provide applications with file access and intra-app communication, respectively. These components are able to service all other apps in the same environment This simplifies maintenance by focusing deployment specific configurations to a single file per environment (1 Config per Env).

Diagram 1-2: Communication Interface

SFTP Server

An sftp file server allows for applications to securely access files from a directory addressed on the file server but stored physically somewhere else. The advantages of accessing files through a server is that extremely complex or changing network mappings can be statically addressed by an app.

The below example uses the atmoz sftp image to create a file server with 2 directories. Directories are defined under the volume tag and match the pattern <Host_Path> : <Container_Path> : <Permission>. The first directory points to /mnt/d/GenStore/sample-data-set/survey-results on the host machine. On the file server this address becomes /home/admin/upload/raw. Likewise the second directory appears under the same parent even though it's location on the host is on a different drive. In addition the ro suffix makes the first directory read only, very useful in safe guarding source files from alteration.

version: '2'

services:
  sftp-server:
    image: atmoz/sftp
    volumes:
        - /mnt/d/GenStore/sample-data-set/survey-results:/home/admin/upload/raw:ro
        - /mnt/f/dept010/data-store/1mk3f5-e5ltb4-4r56yh/01/02:/home/admin/upload/weather-set
    ports:
        - "22:22"
    networks:
      - microservice-network
    command: admin:pass:1001

Messaging Queue

Messaging queues are standard design concepts for microservice applications. Queues allow for applications to be “loosely” connected. Meaning, that each app communicates with the queue instead of with each other. A message queue, in a way, is a running list of small messages. Messages are posted by one app (publishers) and then read by another (subscribers). This example uses an image created by RabbitMQ. RabbitMQ makes connectors for lots of different languages and is efficient for general purpose use cases.
Below is the docker-compose yaml for deploying the RabbitMQ broker.

version: '2'

services:
  rabbit-queue:
    image: rabbitmq:3-management
    hostname: rabbit-1
    environment:
      - RABBITMQ_ERLANG_COOKIE='Key_Link_k1'
    ports:
      - '15672:15672'
      - '5672:5672'
    networks:
      - microservice-network

You can follow a detailed “Hello World” walkthrough in your language of choice from the RabbitMQ site. Below are simple python scripts for a publisher and a subscriber.

def publish ():
    source_path = '/mnt/d/GenStore/sample-data-set/survey-results'
    rbt_srv = 'rabbit-queue'
    trgt_queue = 'new_files'

    try:
        # Get list of file paths to publish
        fAr = os.listdir( source_path )

        print('Connecting to rabbit ',rbt_srv)
        with pika.BlockingConnection(pika.ConnectionParameters(rbt_srv)) as connection:
            channel = connection.channel()
            print('Connected')
            # Create queue if it doesn't
            channel.queue_declare(queue=trgt_queue, durable=True)
            # Clear queue
            channel.queue_purge(queue=trgt_queue)

            i=0
            for f in fAr:
                # Publish file name to queue
                channel.basic_publish(exchange='',
                                    routing_key=trgt_queue,
                                    body=str(f))
                i+=1
            print(" [x] Sent", i,"files to the queue")

            connection.close()
    except Exception as err:
        print("An error occurred while retrieving the file.")
        print(str(err))
        traceback.print_tb(err.__traceback__)
    return

RabbitMQ Publisher rabbit-queue-pub-sub.py

def consume ():
    rbt_srv = 'rabbit-queue'
    src_queue = 'new_files'
    try:
        print(' [-] Connecting to RabbitMQ server',rbt_srv)
        with pika.BlockingConnection(pika.ConnectionParameters(rbt_srv)) as connection:
            channel = connection.channel()

            print(' [+] Connected to RabbitMQ')
            # Declare source queue
            channel.queue_declare(queue=src_queue, durable=True)

            def callback(ch, method, properties, filePath):
                print(" [*] Retrieved file path {0}".format(filePath))
                # Ack to the queue message has been recieved successfuly
                ch.basic_ack(delivery_tag=method.delivery_tag)
            channel.basic_qos(prefetch_count=1)
            channel.basic_consume(queue=src_queue, on_message_callback=callback, auto_ack=False)

            print(' [*] Waiting for messages. To exit press CTRL+C')
            channel.start_consuming()

            connection.close()
    except Exception as err:
        print()
        print("An error occurred while retrieving the file.")
        print(str(err))
        traceback.print_tb(err.__traceback__)
    return

RabbitMQ Subscriber rabbit-queue-pub-sub.py

Complete Script

version: '2'

services:
  sftp-srv:
    image: atmoz/sftp
    volumes:
        - /mnt/d/GenStore/sample-data-set/survey-results:/home/admin/upload/raw:ro
        - /mnt/f/dept010/data-store/1mk3f5-e5ltb4-4r56yh/01/02:/home/admin/upload/weather-set
    ports:
        - "22:22"
    networks:
      - microservice-network
    command: admin:pass:1001

  rabbit-queue:
    image: rabbitmq:3-management
    hostname: rabbit-1
    environment:
      - RABBITMQ_ERLANG_COOKIE='Key_Link_k1'
    ports:
      - '15672:15672'
      - '5672:5672'
    networks:
      - microservice-network
networks:
  microservice-network:
    external: true

Interface hosts: docker-compose.yaml
Python Requirements: requirements.txt