В части 1В этой статье мы рассмотрели параллельную обработку и некоторые сложности, которые она представляет. Важно понимать эти сложности, чтобы оценить инструменты и шаблоны, доступные вам для облегчения бремени. Инструменты, такие как DataRush от Pervasive, помогут вам легче создавать приложения параллельной обработки, которые автоматически масштабируются на многоядерных серверах, не требуя написания сложного многопоточного кода. Во второй части этой статьи рассматривается Pervasive DataRush, его архитектура и модель программирования, а также пример приложения. Это приложение использует DataRush для загрузки больших объемов данных в базу данных, где структура разбивается на части, которые могут выполняться параллельно. Результатом является экспоненциальное ускорение загрузки данных без явной параллельной обработки Java-кода.
Вездесущий Датаруш
Pervasive DataRush — это инфраструктура Java с поддержкой инструментов, которая позволяет быстро создавать высокопараллельные аналитические приложения, полностью использующие современные многоядерные системы. DataRush обрабатывает все проблемы, перечисленные в предыдущих разделах, такие как многопоточность, синхронизация, обнаружение взаимоблокировок, секционирование, постановка в очередь и динамическая настройка времени выполнения на основе ресурсов целевого хоста. Он поддерживает как неявный параллелизм (используя лучшие алгоритмы для решаемой задачи), так и явный параллелизм, где вы можете контролировать, какие аспекты задачи можно сделать параллельными.
Внутри распространенного DataRush
DataRush — это настоящая среда параллельной обработки, построенная на архитектуре на основе потоков данных, которая сочетает горизонтальное и вертикальное разделение с параллелизмом конвейера для использования преимуществ многоядерных систем. Используемый один, любой из этих трех методов полезен; DataRush позволяет объединять их для еще большей эффективности.
Поток данных — это концепция компьютерной науки, в которой операторы обработки данных (небольшие единицы бизнес-логики) связаны между собой через порты, образуя приложение. Операторы считывают данные из входных портов, выполняют их обработку и записывают результаты в выходные порты. DataRush связывает операторские порты с очередями, чтобы позволить операторам работать параллельно.
Приложения DataRush содержат три основных типа компонентов (см. Диаграмму компонентов на рисунке 6):
1.
Процесс : однопоточный скалярный оператор, который реализует некоторую бизнес-логику; базовая единица программирования потока данных в DataRush. Процесс аналогичен подзадаче, как описано в предыдущем примере конвейера;
2.
Сборка : сам оператор, сборка представляет собой многопоточный состав других операторов и настройщиков;
3.
Настройщик : подкомпонент, который динамически конфигурирует, создает или расширяет сборку.
[Img_assist | NID = 2604 | название = | убывание = | ссылка = нет | выравнивание не = средняя | ширина = 502 | Высота = 315]
Правила и отношения, которые управляют этими компонентами, образуют дерево, представляющее законченное приложение DataRush. Сборка всегда является корневым узлом, который содержит другие сборки, процессы и / или настройщики. Когда сборка содержит только другие сборки, это самый верхний корень дерева (см. Рисунок 7).
[Img_assist | NID = 2605 | название = | убывание = | ссылка = нет | выравнивание не = средняя | ширина = 425 | Высота = 353]
Сборка может повторно использовать другие сборки и может содержать произвольное число процессов (для которых вы реализуете бизнес-логику) и классов Customizer (где вы пишете код для изменения определения сборки, когда приложение начинает выполнение). Давайте посмотрим на процесс разработки DataRush подробно.
Разработка компонентов с помощью DataRush
Работа с DataRush требует небольшого изменения мышления для большинства разработчиков. Например, как разработчик Java, ваша первая склонность может заключаться в том, чтобы сделать как можно больше в Java (ловушка, в которую я попал сначала). Однако для того, чтобы воспользоваться преимуществами истинной параллельной обработки без написания множества кода только для решения проблем с потоками, вы должны принять определение сборки DataRush.
Это означает, что для текущей версии вы должны думать с точки зрения сборок DRXML, используя доступные операторы, и ограничивать свой код процессами сборки (для бизнес-логики) и настройщиками, где это применимо. Это хорошо, потому что это означает, что вы будете тратить большую часть своего времени и усилий на реализацию бизнес-логики вместо сложного в отладке кода поддержки многопоточности.
Все начинается со сборки, которую вы определяете в XML в соответствии со спецификацией DRXML. Однако в следующем выпуске Pervasive DataRush весь процесс будет выполнен с использованием подхода на основе Java; вам не нужно работать с DRXML, если вы не хотите. Независимо от того, используете ли вы для этого XML или Java, ваше определение Assembly определяет входные и выходные данные в терминах портов и свойств для динамического управления поведением во время выполнения вашего параллельного приложения (значения свойств соответственно отображаются в файле asupplied .properties). Ваша сборка также может использовать другие сборки, которые вы или другой разработчик уже определили, или один из множества предопределенных операторов данных, предоставленных платформой DataRush (см. Рисунок 8).
Наконец, вы можете определить внешние сборки, которые соединяют выходные порты каждой сборки, на которую ссылаются, с входными портами других. Как только вы сделали это с внешней сборкой, которая сама не определяет порты, у вас есть поток данных, который определяет законченное приложение DataRush, которое готово к выполнению.
[Img_assist | NID = 2606 | название = | убывание = | ссылка = нет | выравнивание не = средняя | ширина = 314 | Высота = 235]
Стрелки в сборке иллюстрируют поток данных (свойства технически не являются частью потока данных). Это определяется с помощью тегов <Link> в файле .drxml определения сборки и может быть легко переставлено (см. Рисунок 9).
[Img_assist | NID = 2607 | название = | убывание = | ссылка = нет | выравнивание не = средняя | ширина = 313 | высота = 235]
В этом примере, хотя в сборке используются те же компоненты, новый поток данных изменяет поведение приложения DataRushapplication без изменения строки кода Java.
Имейте в виду, что термин оператор обобщает другие сборки и процессы сборки. Это позволяет создавать базовые строительные блоки кода параллельной обработки и использовать их рекурсивно, чтобы легко собирать новые параллельные приложения без необходимости переделки. Одно определение сборки может содержать несколько экземпляров процесса и настройщика для дальнейшего разбиения обработки на более мелкие и простые этапы. Таким образом, используя платформу DataRush, вы получаете преимущества как горизонтального, так и вертикального разделения для параллельной обработки, максимально увеличивая повторное использование кода.
На данный момент у вас, вероятно, есть несколько вопросов, таких как:
- Как осуществляется передача данных из одного порта сборки в другой?
- Сколько потоков используется для управления выполнением моих сборок?
- Когда частично доступные данные передаются?
К счастью, вам не нужно беспокоиться об этих деталях. DataRush поддерживает в памяти, блокирует, очереди между портами, чтобы гарантировать, что данные доступны для сборок, когда они им нужны, даже если они выполняют постоянные скорости. Кроме того, DataRush запускает правильное количество потоков в соответствии с количеством заданных вами процессов сборки, с некоторой настройкой, выполняемой в соответствии с количеством доступных процессоров / ядер.
Библиотеки, инструменты и среда разработки DataRush
DataRush в основном состоит из набора библиотек и простых файлов сценариев, чтобы связать их вместе. Библиотеки содержат код для механизма исполнения DataRush, его API программирования, предоставленный набор операторов, которые можно повторно использовать в ваших собственных приложениях DataRush, и некоторый внутренний служебный код.
Три файла сценария в каталоге bin вашей установки DataRush определяют три основных инструмента, которые вы будете использовать:
- dra : Это ассемблер DataRush. Он принимает определения сборки (файлы .drxml) вместе со ссылками на файлы .java (для реализаций процессов и настройщиков) и компилирует их в исполняемое приложение DataRush. Инструмент dra вызовет компилятор Java для ссылочного кода Java.
- dre: This is the DataRush execution engine. Use this tool to execute a DataRush application once it has been assembled with the dra tool.
- dri: This is the DataRush inspector tool. You can use this tool to inspect the internals of other DataRush applications and Assemblies. Use of this tool is not needed during the code, compile, run phases of a DataRush application, but is provided for your convenience.
Also included is a DataRush Eclipse plug-in that allows you to visually inspect your Assembly definitions and DataRush applications as you define them. Of course, you can also easily create new DataRush projects, Assembly definition files, process and customizer Java classes, and other filesneeded for DataRush applications, from within Eclipse (see Figure 10).
[img_assist|nid=2608|title=|desc=|link=none|align=middle|width=604|height=640]
As you’re writing code or defining assemblies, you compileand debug your DataRush application like you would a plain Java application within Eclipse, with full tools support. For example, you can create application launch configurations and run DataRush projects within Eclipse like any other Java application. Assembly definitions, data flows, and DataRush application runtime output are all visible without leaving Eclipse (see Figure11).
[img_assist|nid=2611|title=|desc=|link=none|align=middle|width=789|height=534]
Of course, all of this functionality (minus the visuals) is available through the command-line tools, dri (DataRush inspector), dra (DataRush assembler), and dre (DataRush engine), outside of the Eclipse environment.
High-Speed, Parallel, Database Loading with DataRush
To illustrate the power of DataRush, let’s explore a sampleapplication that simulates an RSS news feed aggregator. In this application, the XML files that are received from different RSS feeds are to be read andparsed. Then, the pertinent fields are written to a comma-delimited file (CSV)for bulk import into a database. In this sample, the database will be MySQL,and the mysqlimport tool will be usedfor bulk loading. Of course, other databases can be used just as easily.
The files from the various feeds reside in separate directories, where each directory can contain zero or more XML files. The generated CSV files are then available to be bulk loaded. Each step of the process (reading the XML files, parsing them, writing the CSV files, and calling the bulk import tool) can be done in parallel, and even the tasks within can benefit from partitioning and pipelining for even more benefit. DataRush will be used to achieve this with minimal coding.
The combination of DataRush to achieve maximum parallelization on your multi-core server, with the high-speed bulk loader tools available with most database engines (i.e. MySQL, Oracle), results in a high-speed, scalable, database loader application with no thread-specific code needed. All of the multi-threading is handled automatically by DataRush.
For this particular project, I utilized DRXML for significant portions of the process. However, Pervasive DataRush expects to move to an all-Java approach with its Beta 3 release later this year. Once released, Beta 3 will be available as a free download at www.pervasivedatarush.com. For now, let’s take a look at the sample DRXML-based application in detail.
DataRush Components
The diagram in Figure 12 shows the assemblies and data flow for the application just described. Since the DataLoader Assembly contains no import or output ports, it defines the DataRush application. It’s also a composite Assembly that contains custom assemblies for the other phases of work. Here is a description of each:
- DataLoader: This is the application Assembly that starts the whole process through these two main steps:
- Pass the base path property to the DirectoryCrawler Assembly to begin file processing.
- A customizer iterates through all subdirectories, and modifies the DataLoader assembly composition (via the DataRush Java APIs) to add instances of each Assembly (DirectoryCrawler, RSSParser, and ArticleImporter) for each subdirectory found. In this sense, the Assembly is recursive.
- DirectoryCrawler: This Assembly is given a base directory, searches for all XML files in the directory, parses their contents using a Document Object Model (DOM) parser, and sends each file’s DOM through the output port as an Object.
- RSSParser: This Assembly takes an XML DOM object as input, traverses the DOM data hierarchy, generates a CSV file with the pertinent article data, and sends the resulting filename via the output port.
- ArticleImporter: This Assembly is given an importer tool to use as a property, along with a CSV file from the input port, and calls the import tool with that filename.
[img_assist|nid=2612|title=|desc=|link=none|align=middle|width=400|height=278]
The DataCrawler Assembly contains a customizer, but no process class. The other assemblies each contain a process class to perform their related business processing. The definition for the DataLoader Assembly is shown in Listing 1. The sections of interest include the <Customizer> entry, and the three <Assembly> entries.
The customizer entry references a customizer class in the type attribute, the stage at which to invoke the customizer, and variables that are set. In this case, the customizer will be invoked at composition time, which allows the class to modify the actual Assembly file before it’s compiled. The base directory path to search for XML files and subdirectories, and theimport tool are provided—via <Set>elements—to the customizer, seeded from the Assembly properties:
<Customizerstage="composition" type="DataLoaderCustomizer">
<Set target="path"source="path" />
<Set target="importer"source="importer" />
</Customizer>
Next, an instance of the DirectoryCrawler Assembly isdeclared, and the base path property is provided:
<Assemblyinstance="directoryCrawler" type="DirectoryCrawler">
<Set target="path"source="path" />
</Assembly>
This Assembly instance will search for all XML files in thepath, and send each filename via its output port. Next, the DataLoader Assembly declares an instance of the RSSParser Assembly and links its input port to the output from DirectoryCrawler via the <Link> element. The drxml is verbose and intuitive enough to understand when you read through it:
<Assemblyinstance="rssParser" type="RSSParser">
<Linkinstance="directoryCrawler"
source="domOutput"
target="domInput" />
</Assembly>
Next, an instance of the ArticleImporter Assembly is referenced. In this case, its input port (for the CSV file to be loaded) is linked to the output of RSSParser, and the importer property is passed in:
<Assemblyinstance="articleImporter" type="ArticleImporter">
<Set target="importer"source="importer" />
<Link instance="rssParser"
source="filenameOutput"
target="csvFilenameInput" />
</Assembly>
Now let’s take a look at the RSSParser Assembly since it declares both an input and output port, two properties, and a process class (see Listing 2). Ports and properties are declared at the top of the drxml file, within the <Contracts> section. Again, thanks to the verbosity of XML, you can read this and understand what the intent is:
<Contract>
<Inputs>
<Portname="domInput" type="object"/>
</Inputs>
<Outputs>
<Portname="filenameOutput" type="string"/
</Outputs>
<Properties>
<Propertyname="fileName" type="string" default="null"/>
<Propertyname="charsetName" type="string"default="ISO-8859-1" >
</Properties>
</Contract>
Next, the Assembly’s process class is declared in the<Process> element. The process class is set with the type attribute, and the instance name is defined. Also, the class’s internal variable, input, is linked to the input port via a <Link> element:
<Composition>
<Processinstance="rssParserProc" type="RSSParserProcess">
<Linksource="domInput" target="input" />
</Process>
<Linkinstance="rssParserProc"
source="filenameOutput"
target="filenameOutput"/>
</Composition>
Notice that the output port, filenameOutput, is linked to the process’s internal variable, also named filenameOutput. For the links to succeed, you need to define public getter and setter methods in the process class so the DataRush framework to access the variables (see Listing 3).
Note: Since the application depends on files from multiple directories, with a dataflow per directory, parallel processing only occurs when files are distributed across the directories. It was built this way for the purposes of this article. However, this can limit the degree or parallel processing that will occur if files aren’t evenly distributed. To remedy this in a real-world solution, the DirectoryCrawler Assembly should feed a queue that is read by a set of paired RSSParser – ArticleImporter Assemblies that execute in parallel.
Task Partitioning
Before execution begins, the DataLoader customizer class (see Listing 4) inserts instances of each of the custom assemblies for each subdirectory it finds in the base directory provided. As a result, each directory is processed in parallel, and data pipelining takes place throughout the XML file parsing and CSV file generation steps to achieve parallelization for the files within each directory.
All of the files within a single directory are parsed, and the important article data within them is written to a single CSV file. This happens in parallel to the assemblies working on the XML files within the other subdirectories. The data is pipelined in parallel between the assemblies working across all of the directories. Finally, for each CSV file created, DataRush invokes the database bulk loader via a command line script.
For directories with many, and larger, files, parsing may continue while the assemblies processing less populated directories finish creating their CSV files and begin the import process. The final result is a balance between CPU-intense parsing, IO operations for file reads and writes, and database bulk loading, all of which is made possible thanks to DataRush, without the need to write special multi-threaded code.
Running the Sample Application
To run the sample application, you can load the project in Eclipse after installing both the DataRush SDK and the Eclipse plug-in for DataRush development. You can find the source, article data, and other support files for the sample application here. You’ll also need to create a database named newsdata with a tablenamed article. Here is the SQLstatement that creates the table with the proper columns in MySQL:
DROPTABLE IF EXISTS `newsdata`.`article`;
CREATETABLE `newsdata`.`article` (
`TITLE` varchar(80) NOT NULL,
`ID` varchar(45) NOT NULL,
`DATE` varchar(45) default NULL,
`SUMARRY` varchar(1200) default NULL,
`URL` varchar(45) default NULL,
PRIMARY KEY (`ID`)
)ENGINE=MyISAM DEFAULT CHARSET=latin1;
You can then execute the sample application, which works on the XML data within the subdirectories in the data directory. Alternatively, you can execute the following command from the sample application’s root directory, after you ensure the DataRush command-line tools are in your path:
$ dre -cp bin -pfDataLoader.properties DataLoader
Remember that you’ll need Java 6 to run DataRush. The output from the execution should appear similar to that shown in Listing 5. Running the application on a machine with multiple cores should run magnitudes faster than with one cpu/core. With the data packaged with the sample, which includes about 50,000 articles (some real, most generated), I observed the following numbers:
- 2.8GHz single-core CPU: 8.234 seconds for import
- 2.4GHz dual-core CPU: 3.167 seconds for import
With these numbers, you can see that parallel processing improves the numbers even though the single-core CPU I ran the application on runs at a higher clock-rate. In fact, despite the slower clock rate, the same DataRush application run more than twice as fast on the dual-core CPU. The real magic here is that DataRush allowed me to take advantage of the additional core(s) without writing any additional code.
Key Learning and Takeaways
In this article, you’ve seen that developing parallel processing applications that truly take advantage of today’s multi-core servers requires more than writing multi-threaded code. It requires a framework that performs the multi-threading work for you, and allows for task partitioning, and data pipelining, to ensure that even seemingly single-threaded tasks get parallelized.
Further, you’ve seen how Pervasive provides such a framework with its DataRush product, which allows Java developers to concentrate on writing reusable business logic in Java, while providing a parallel processingruntime environment that scales to your multi-core hardware. What’s more, withits focus on tools, and complete integration with Eclipse, DataRush takes almost all of the complexity out of writing and running parallel-processing applications for massive data crunching.