Merge dvou tabulek v Pythonu

Aktuálně studuju na Coursera kurz Introduction to Data Science in Python a jak se to někdy hezky sejde, naskytla se mi v práci možnost to rovnou použít v praxi.

Rozhodně nejde o nic světoborného - prostě potřebuju dát dohromady dvě tabulky, trochu pošolichat data a udělat hierarchický index. Asi by to šlo udělat i v něčem jiném a možná jednodušeji. Ale Python mám rád a je to zábava si trochu zablbnout.

Následující zápisek je zdokumentování postupu, který by se mi mohl někdy v budoucnu hodit. Pokud jste někdo větší borec v Pythonu - tj. nepíšete v něm jen jednou za pár let, jako já - budu rád, když mi poradíte nějaké zlepšení.

O co jde?

Určitě jste se s tím už setkali. Máte určitý nástroj, od kterého byste čekali celkem jednoduchou funkcionalitu a on ji, z nějakého důvodu, nemá. Takže skončíte u exportu dat a nějaké externí úpravy.

To je náš výchozí bod - máme dvě vyexportované tabulky, formát souboru v tomhle případě nehraje roli. Může to být CSV, Excel atd. Ty naše dvě vypadají takhle:

Tabulka CSD.csv

Tabulka RQS.csv

Co je pro nás podstatné, obě tabulky mají vzájemnou vazbu přes sloupec "Related issues", který odkazuje na identifikátor záznamu v druhé tabulce (sloupec "#"). Co není na první pohled zřejmé, záznamy v tabulce CSD (zelené záhlaví) mají vazbu 1:N na záznamy v tabulce RQS (červené záhlaví). Pro úplnost, vazba je obousměrná, nás ale zajímá jen směr CSD -> RQS.

No a potřebovali bychom z toho dostat tabulku, která vypadá takto (barevné rozlišení je pro přehlednost, který data kam patřily):

Výsledná tabulka matrix.xlsx

Povšimněte si v prvních dvou sloupcích hierarchického indexu ("CSD ID", "RQS ID"), který vyjadřuje vazbu 1:N.

Jak na to?

To, kolem čeho se točí výše zmíněný kurz a co jsem pro manipulaci s tabulkami použil, je knihovna pandas - open source nástroj pro datové struktury a datovou analýzu. Na svých stránkách píšou, že je easy-to-use a opravdu se s tím pěkně pracuje.

Základním elementem v pandas je DataFrame, ekvivalent dvourozměrného pole (se spoustou vychytávek). První krok je, převést naše tabulky do DataFrame. pandas mají spoustu možností, jak načíst externí data, my použijeme metodu read_csv:

import pandas as pd

csd = pd.read_csv('CSD.csv')
rqs = pd.read_csv('RQS.csv')

Zpracovávaná tabulka může být rozsáhlá, co do počtu sloupců, a protože výpis DataFramu na konzoli se defaultně zalamuje, může se hodit příkaz pro vypnutí této možnosti:

pd.set_option('display.expand_frame_repr', False)

Data máme načtená, můžeme je začít upravovat. První na řadě je tabulka CSD. Potřebujeme udělat následující úpravy:

• Rozdělit data ze sloupce "Subject" do dvou samostatných sloupců pomocí oddělovače " > ".
• Smazat sloupce, které v cílové tabulce nepotřebujeme.
• Přejmenovat sloupce, aby v cílové tabulce dávali větší smysl.

csd[['CSD ID', 'Contract chapter']] = csd['Subject'].str.split(
        ' > ', expand=True)
csd.drop(['Category', 'Subject',
    'Assigned To', 'Target version',
    'Release', 'Related issues'], axis=1, inplace=True)
csd = csd.rename(columns={'Status': 'CSD status'})

Navíc uděláme jednu operaci, kterou budeme potřebovat později - přidáme si nový sloupec "numeric ID", podle kterého cílovou tabulku později setřídíme.

csd['numeric ID'] = csd['CSD ID'].str.replace('-', '').map(int)

Obdobné úpravy uděláme na tabulce RQS a můžeme se pustit do spojování. pandas nabízejí metodu merge, která funguje obdobně jako SQL join.

matrix = csd.merge(rqs, left_on='#',
        right_on='Related issues', how='left')

Dalším krokem je vytvoření hierarchického indexu:

matrix.set_index(['CSD ID', 'RQS ID'], inplace=True)

Následně setřídíme novou tabulku podle sloupce "numeric ID", který jsme si dočasně přidali v rámci úprav tabulky CSD. Sloupec po setřídění odstraníme. Smyslem téhle eskapády je setřídit tabulku numericky podle sloupce, kde jsou string hodnoty.

matrix = matrix.sort_values(by='numeric ID')
matrix.drop('numeric ID', axis=1, inplace=True)

Teď si ještě seřadíme sloupce, abychom se v cílové tabulce dobře orientovali:

cols = ['Contract chapter', 'CSD status',
        'Redmine ID', 'RQS description',
        'RQS status', 'Related issues']
matrix = matrix[cols]

A pozor! Finální příkaz... zapíšeme do Excelu, resp. jiného výstupního formátu. Máme hotovo.

matrix.to_excel('matrix.xlsx')

Kompletní skript

Celý skript je k dispozici na Bitbucketu jako snippet: matrix.py.

Jak nainstalovat pandas?

Nejjednodušší způsob, jak nainstalovat pandas a další spřízněné knihovny (třeba NumPy a dokonce i samotný Python) je package manager Miniconda. Stačí stáhnout instalátor pro váš operační systém a pak nainstalovat pandas příkazem:

conda install pandas

Vytvoření JMS Bridge na WebLogicu

Messaging bridge je šikovné řešení, pokud potřebujeme distribuovat zprávy mezi několika messaging systémy. Potřeboval jsem vytvořit JMS bridge na WebLogicu a protože jsem si oblíbil WebLogic Scripting Tool (WLST), tak jsem si napsal skript.

Nicméně dnes, vás milí čtenáři, nechci odfláknout pouhým WLST skriptem, ale podíváme se na téma trochu zeširoka. Prvně si zasadíme JMS bridge do kontextu Enterprise Integration Patterns (EIP), pak se podíváme, jak jde bridge naklikat ve WebLogic konzoli. A samozřejmě vás neochudím o to WLST :-)

Messaging Bridge

Nebyl bych to já, kdybych se nevytasil s nějakým patternem. Tak tady je - Messaging Bridge, jak je definován v EIP. (Věty kurzivou jsou citacemi z knihy Enterprise Integration Patterns: Designing, Building, and Deploying Messaging Solutions.)

Messaging Bridge řeší následující problém: "How can multiple messaging systems be connected so that messages available on one are also available on the others?" Integrační vzory řeší problémy obecně. V tomto případě jde o to, jak dostat zprávy z jednoho systému na jiný. Např. jak dostat zprávy z WebSphere MQ na MSMQ. Nebo JMS. Nebo TIBCO Randevouz. Nebo... Jasný, ne? Odkudkoliv kamkoliv.

Messaging Bridge tedy je "a way for messages on one messaging system that are of interest to applications on another messaging system to be made available on the second messaging system as well."

Protože většinou neexistuje způsob, jak propojit dva různé messaging systémy, propojují se jednotlivé, odpovídající kanály na daných systémech. "The Messaging Bridge is a set of Channel Adapters ... where each pair of adapters connects a pair of corresponding channels. The bridge acts as a map from one set of channels to the other and transforms the message format of one system to the other."

Vzor Messaging Bridge (zdroj EIP)

JMS Bridge

JMS bridge je speciální variantou Messaging bridge, který má několik omezení, nebo spíše možná zjednodušení. Tři hlavní jsou:

• Propojuje pouze JMS systémy (různých dodavatelů).
• Zprávy netransformuje (protože pracuje pouze s JMS zprávami).
• Zprávy z jednoho (zdrojového) systému přeposílá na jiný (cílový). Čili nečiní je pouze dostupnými, ale provádí routing/forwarding.

JMS bridge běžně poskytují všichni dodavatelé JMS systémů, např.:

• HornetQ (JBoss)
• ActiveMQ (Apache)
• WebLogic (Oracle)

Zajímavým atributem u JMS bridge je Quality of Service (QoS), který říká, v jakém modu budou zprávy přeposílány:

• At most once - nanejvýš jednou. Toto je nejbenevolentnější mod. Zpráva buď dorazí (maximálně jednou), anebo taky ne. A nám to tak vyhovuje.
• Duplicates OK - duplicity jsou v pořádku. V tomto modu jsou zprávy potvrzeny, že dorazily na cílový systém. Může se stát, že stejná zpráva nám dorazí ve více instancích, ale to je v pořádku - řekli jsme přece, duplicity jsou OK. Výhodou je, že se žádná zpráva neztratí.
• Exactly once - přesně jednou. Toto je nejvíc striktní mod, který nám nejen zaručuje, že každá zpráva dorazí na cílový systém, ale taky že tam dorazí právě jednou. Takový závazek není jen tak a zajistí nám ho JTA, čili distribuovaná transakce.

WebLogic JMS Bridge

Vytvoření JMS bridge na WebLogicu je otázkou chvilky, stačí mít připraveny správné ingredience. Co budeme potřebovat?

• URL (a credentials) zdrojového serveru,
• URL (a credentials) cílového serveru,
• JNDI JMS adapteru (transakční, nebo netransakční)
• eis.jms.WLSConnectionFactoryJNDIXA
• eis.jms.WSLConnectionFactoryJNDINoTX
• JNDI Connection Factory,
• JNDI JMS destinace (fronty),
• Quality of Service
• Exactly-once
• Duplicate-okay
• Atmost-once

Pak už stačí jen naházet to do hrnce, zamíchat, podusit... a proklikat se průvodcem. Protože těch obrazovek ve wizardu je zbytečně moc, ukážu jenom screenshoty konečného stavu.

Pokud chceme použít transakční mod Exactly once, je dobré ještě před vytvářením bridge deployovat transakční adaptér jms-xa-adp.rar - vyhneme se tak zbytečným restartům (adapteru či WebLogicu). Adapter najdeme mezi knihovnami WebLogicu a nasazujeme ho jako aplikaci.

Nasazený resource adapter jms-xa-adp.rar (ve struktuře Deployments)

Bridge se sestává ze dvou destinací (zdrojové a cílové):

Definice JMS (source) destination

a samotného bridge:

Definice JMS bridge

Že nám bridge funguje, poznáme podle jeho stavu (záložka Monitoring) a taky doporučuju si nějakou zprávu cvičně přeposlat. Světe div se, ale transakce můžou zlobit.

Stavy JMS bridgů (záložka Monitoring)

WLST

No a máme tady velké finále, ke kterému jsem nenápadně, ale cílevědomě směřoval. Takže tady je: WLST v celé své pythoní kráse :-)

#
# properties
#
user = 'weblogic'
password = 'welcome1'
server = 't3://sandman:7001'
# source credentials
sourceJmsUser = 'weblogic'
sourceJmsPassword = 'welcome1'
sourceURL = 't3://sandman:7003'
# target credentials
targetJmsUser = 'weblogic'
targetJmsPassword = 'welcome1'
targetURL = 't3://sandman:7004'
# JMS servers
servers = ['SourceServer']
targets = []
# queues
queues = [
    'EVM_EVE_CMS',
    'EVM_EVE_FC',
    'EVM_EVE_GEN',
    'EVM_EVE_PTS',
    'NTF_PARTY',
    'NTF_PRODUCT']
adapterJNDI = 'eis.jms.WLSConnectionFactoryJNDIXA'
connectionFactory = 'jms/sws/SWSConnectionFactory'
qualityOfService = 'Exactly-once'
jndiPrefix = 'jms/sws/'

# connection
connect(user, password, server)

# edit mode
edit()
startEdit()

# get targets
for server in servers:
    targets.append(getMBean('/Servers/' + server))
print '\nTargets:', targets, '\n'

#
# create bridges
#
print '\n### Create bridges:\n'
for queue in queues:
    bridgeName = queue + '_bridge'
    sourceName = queue + '_source'
    targetName = queue + '_target'
    # delete an old bridge
    ref = getMBean('/MessagingBridges/' + bridgeName)
    if ref != None:
        cd('/MessagingBridges')
        delete(bridgeName, 'MessagingBridge')
    # delete an old source destination
    ref = getMBean('/JMSBridgeDestinations/' + sourceName)
    if ref != None:
        cd('/JMSBridgeDestinations')
        delete(sourceName, 'JMSBridgeDestination')
    # delete an old target destination
    ref = getMBean('/JMSBridgeDestinations/' + targetName)
    if ref != None:
        cd('/JMSBridgeDestinations')
        delete(targetName, 'JMSBridgeDestination')
    # create a new source destination
    cd('/')
    cmo.createJMSBridgeDestination(sourceName)
    cd('/JMSBridgeDestinations/' + sourceName)
    cmo.setAdapterJNDIName(adapterJNDI)
    cmo.setConnectionFactoryJNDIName(connectionFactory)
    cmo.setConnectionURL(sourceURL)
    cmo.setDestinationJNDIName(jndiPrefix + queue)
    cmo.setUserName(sourceJmsUser)
    cmo.setUserPassword(sourceJmsPassword)
    print 'Source:', cmo
    # create a new target destination
    cd('/')
    cmo.createJMSBridgeDestination(targetName)
    cd('/JMSBridgeDestinations/' + targetName)
    cmo.setAdapterJNDIName(adapterJNDI)
    cmo.setConnectionFactoryJNDIName(connectionFactory)
    cmo.setConnectionURL(targetURL)
    cmo.setDestinationJNDIName(jndiPrefix + queue)
    cmo.setUserName(targetJmsUser)
    cmo.setUserPassword(targetJmsPassword)
    print 'Target:', cmo
    # create a new bridge
    cd('/')
    cmo.createMessagingBridge(bridgeName)
    cd('/MessagingBridges/' + bridgeName)
    cmo.setSourceDestination(getMBean('/JMSBridgeDestinations/' + sourceName))
    cmo.setTargetDestination(getMBean('/JMSBridgeDestinations/' + targetName))
    cmo.setStarted(true)
    cmo.setQualityOfService(qualityOfService)
    cmo.setTargets(targets)
    print 'Bridge:', cmo, '\n'

# save and finish
save()
activate()
disconnect()
exit()

Související články

• Vytvoření WebLogic Distributed Queue pomocí WLST
• Vytvoření JDBC datasource na WebLogicu pomocí WLST
• Enterprise integrace, messaging
• Dead Letter Channel nebo Invalid Message Channel? Toť otázka

Vytvoření JDBC datasource na WebLogicu pomocí WLST

Dneska to bude jen taková variace na minulé téma (vytvoření JMS zdrojů pomocí WLST), aneb jak na WebLogicu vytvořit JDBC datasource pomocí skriptovacího nástroje WLST (WebLogic Scripting Tool).

Podstatné věci o WLST jsem zmínil v uvedeném postu, takže je tady už nebudu opakovat a eventuálně vás pro doplňující informace odkazuji tam.

#
# properties
#
user = 'weblogic'
password = '<password>'
server = 't3://<host>:7001'
dsName = '<datasource>'
dsPath = '/JDBCSystemResources/' + dsName + '/JDBCResource/' + dsName
clusterName = 'soa_cluster'

# connection
connect(user, password, server)

# edit mode
edit()
startEdit()

#
# delete an old JDBC resource
#
if cmo.lookupJDBCSystemResource(dsName):
    delete(dsName, 'JDBCSystemResource')

#
# create JDBC resource
#
cmo.createJDBCSystemResource(dsName)
# set datasource name
cd(dsPath)
cmo.setName(dsName)
# set JNDI name
cd(dsPath + '/JDBCDataSourceParams/' + dsName)
cmo.setJNDINames(['jdbc/' + dsName])
# set driver
cd(dsPath + '/JDBCDriverParams/' + dsName)
cmo.setDriverName('oracle.jdbc.xa.client.OracleXADataSource')
cmo.setPassword('<password>')
cmo.setUrl('jdbc:oracle:thin:@<host>:1521:<SID>')
# set username
cd(dsPath + '/JDBCDriverParams/' + dsName + '/Properties/' + dsName)
cmo.createProperty('user')
cd(dsPath + '/JDBCDriverParams/' + dsName + '/Properties/' + dsName + '/Properties/user')
cmo.setValue('<user>')
# set targets
cd('/JDBCSystemResources/' + dsName)
cmo.setTargets([getMBean('/Clusters/' + clusterName)])

# save and finish
save()
activate()
disconnect()
exit()

Související články

• Vytvoření WebLogic Distributed Queue pomocí WLST
• Odstranění metadat z MDS

Vytvoření WebLogic Distributed Queue pomocí WLST

Pokud nějaká aplikace používá JMS zdroje, bývají  tyto zpravidla externí. (Výjimkou je JMS broker embeddovaný uvnitř aplikace.) Tyto externí zdroje bývají často vytvořeny na aplikačním serveru, ve kterém je většinou JMS server už obsažen. Pokud naše aplikace používá např. JMS fronty, musí je "někdo" na daném JMS serveru vytvořit. Ten někdo je na vývojovém a někdy i testovacím prostředí vývojář, na dalších prostředích už to bývá administrátor.

Podle daného aplikačního serveru se JMS zdroje dají buď naklikat v nějaké administrátorské konzoli, nebo je potřeba poeditovat/vytvořit nějaké konfigurační soubory. Třetí možností je tyto  zdroje nějakým nástrojem naskriptovat. V případě aplikačního serveru WebLogic je takovým nástrojem WebLogic Scripting Tool (WLST).

Jak už jsem psal v minlém zápisku o mazání dat z MDS, WLST je v Jythonu napsaný nástroj pro správu WebLogic serveru, který funguje ve dvou režimech - offline a online. V online režimu se WLST připojuje k běžícímu WebLogicu a operuje nad stromem jeho MBeans. Pro správu JMS zdrojů je potřeba používat WLST online.

Distributed Queue

Věc, kterou jsem potřeboval vyřešit, bylo vytvoření distribuované fronty ve WebLogicovém clusteru. Cluster byl velmi jednoduchý - admin server a dva nody (managed servery). Vzhledem ke clusteru bylo potřeba vytvořit logickou frontu, která by měla stejný JNDI (jako na vývojovém prostředí s jedním nodem) a zastřešovala by fronty na jednotlivých nodech. Na WebLogicu je toto řešeno distribuovanými destinacemi (queue/topic). Pokud si to neumíte představit, distribuovaná fronta funguje v podstatě jako klasický load balancer.

Distribuovaná fronta (WebLogic Administration Console)

Členové distribuované destinace (WebLogic Administration Console)

WLST a JMS

Použití WLST je po krátké praxi poměrně intuitivní a jednoduché. Zpočátku může dělat problém se orientovat ve struktuře (stromech) MBean. V tomto může napomoci celkem slušná dokumentace: navigace MBeans, Javadoc a MBean reference. Pak už stačí jenom lehké základy Pythonu.

V následujícím skriptu je několik věcí, které můžou trochu ztížit čtení/pochopení skriptu, takže ještě kratičká legenda:

• Target. Cílové umístění zdroje, nebo aplikace, např. server, cluster, JMS server ad. V případě deploymentu (aplikace) tím říkám, na které nody/clustery chci aplikaci nasadit.
• SubDeployment. Mechanismus pro seskupení a umístění JMS zdrojů. V rámci subdeploymentu říkám, na které cíle (targets) chci dané zdroje nasadit. Může to být libovolná kombinace, nebo podmnožina.
• cmo. Proměnná, která reprezentuje "aktuálně spravovaný objekt" (current management object). Je to vlastně MBeana, která má momentálně "focus". Tak, jak se prochází stromem MBean, tak se tato proměnná automaticky mění.

# properties
user = 'weblogic'
password = '<password>'
server = 't3://<host>:7001'
subDeploymentName = 'EVMJMSServers'
queuePath = 'JMSResource/SOAJMSModule/UniformDistributedQueues/'
queueName = 'EVM_DLQ'
jndiPrefix = 'jms/b2b/'
loadBalancing = 'Round-Robin'

# connection
connect(user, password, server)

# edit mode
edit()
startEdit()

# get targets
s1 = getMBean('/JMSServers/SOAJMSServer_auto_1')
s2 = getMBean('/JMSServers/SOAJMSServer_auto_2')

#
# create a SubDeployment
#
cd('/JMSSystemResources/SOAJMSModule')
# delete an old SubDeployment
if cmo.lookupSubDeployment(subDeploymentName):
    delete(subDeploymentName, 'SubDeployment')
# create a new SubDeployment
cmo.createSubDeployment(subDeploymentName)
subDeployment = cmo.lookupSubDeployment(subDeploymentName)
subDeployment.setTargets([s1, s2])

#
# create a ldistributed queue
#
resource = cmo.getJMSResource()
# delete an old queue
ref = getMBean(queuePath + queueName)
if ref != None:
    cd('JMSResource/SOAJMSModule')
    delete(queueName, 'UniformDistributedQueue')
# create a new queue
resource.createUniformDistributedQueue(queueName)
distributedQueue = resource.lookupUniformDistributedQueue(queueName)
distributedQueue.setJNDIName(jndiPrefix + queueName)
distributedQueue.setLoadBalancingPolicy(loadBalancing)
distributedQueue.setSubDeploymentName(subDeploymentName)

# save and finish
save()
activate()
disconnect()
exit()

Související články

• Odstranění metadat z MDS

Odstranění metadat z MDS

Dneska to bude jedna praktická. Aneb jak v SOA Suite smazat z MDS (MetaData Store) nějaká metadata, soubory nebo adresář. Práce s MDS není z počátku úplně intuitivní, dokumentace není úplně jednoduše k nalezení a informace aby člověk vyškrabával z různých blogů a fór.

Tak, jak na to. Možnosti jsou dvě. Jednak použít WLST, nebo Ant skript, který je součástí SOA Suite (ten ale umožní pouze smazání adresáře).

WLST

WebLogic Scripting Tool (WLST) je zajímavý, v Jythonu napsaný, nástroj na administraci WebLogic serveru. Lze s ním pracovat dvěma způsoby: offline a online. Bohužel, s ohledem na MDS, je potřeba pro mazání souborů použít online způsob a pro smazání adresáře naopak offline (a také se používají dvě různé funkce). WLST zpravidla najdeme v adresáři <ORACLE_HOME>/oracle_common/common/bin.

Smazání souboru

Pro smazání jednoho a více souborů lze použít příkaz deleteMetadata. Vzhledem k tomu, že se tento příkaz používá online, je potřeba se prvně připojit k WebLogic administračnímu serveru. A po zadání mazacího příkazu se zase odpojit.

connect('weblogic', '<password>', 't3://<AdminServer>:7001')
deleteMetadata(
        application='soa-infra',
        server='soa_server1',
        docs='/apps/<pathToFile>/DeadLetterQueue-v1.0.wsdl')
disconnect()
exit()

Smazání adresáře

Pro smazání adresáře slouží offline příkaz sca_removeSharedData. Problém s tímhle příkazem je, že není součástí standardních WLST modulů, ale je součástí instalace SOA Suite. Proto aby fungoval, je potřeba spustiti WLST z odpovídajícího adresáře: <ORACLE_HOME>/Oracle_SOA1/common/bin.

Dalším rozdílem je, že se nezadává cesta k administrativnímu serveru, ale ke spravovanému (managed) serveru, na kterém MDS běží.

sca_removeSharedData(
    'http://<ManagedServer>:8001',
    'public/evm/iface/DeadLetterQueue-v1',
    'weblogic', '<password>')

Ant

Součástí instalace vývojového prostředí pro SOA Suite (JDeveloper) je i sada Ant skriptů mmj. pro buildování, (unit) testování a deployment. A také pro smazání adresáře z MDS. Tímto způsobem nejde mazat jednotlivé soubory.

Soubory se dají najít v adresáři <ORACLE_HOME>/jdeveloper/bin.

ant -f ant-sca-deploy.xml removeSharedData
    -DserverURL=<ManagedServer>
    -DfolderName=<pathToFolder>
    -Duser=<user>
    -Dpassword=<password>

Související články

• Java a fault handling policies v Oracle SOA Suite
• Custom XSLT funkce v Oracle SOA Suite