Staging Large Dataset from SQL DB to Hazelcast

[Anatoly Polinsky]

Dear Fellow Hazelcasters,

I would like to gather opinions on "How do I do this the most
effective way"...

GIVEN:

SQL Sever 2005 DB with a table that potentially has tens of
millions of rows that need to be processed
This table has a composite PK of two VARCHAR fields ( so no
mathematically "iteratable" key )
This is a client table that cannot be in any way changed
( can't add an IDENTITY, natural PK, etc... )

LOOKING FOR:

Reading all the rows that needs processing into a Hazelcast's
List, which will naturally give me a numeric PK ( list's index )

NEED YOUR OPINION ON:

What would be the most efficient way to stage such a List from
DB, given the fact that there can be tens of millions of rows that
needs processing.

Thank you,
/Anatoly

[Talip Ozturk]

Hazelcast's List doesn't support index so you cannot use
List.get(index). But the good part is that you are saying you already
have composite PK, then use Map.

class CompositeKey {
String fieldOne;
String fieldTwo;
}

map.put(new CompositeKey(fieldOne, fieldTwo), RowObject);

Besides these:
1. Make sure that when your cluster starts, each member has same
number of partitions. Otherwise it won't fit in the memory of the node
holding/owning most of the partitions.
2. Make sure you have enough machine (memory). Calculate each entries
cost = (backup-count + 1) X (key.binary.size + value.binary.size +
440bytes).

http://twitter.com/oztalip

> --
> You received this message because you are subscribed to the Google Groups "Hazelcast" group.
> To post to this group, send email to haze...@googlegroups.com.
> To unsubscribe from this group, send email to hazelcast+...@googlegroups.com.
> For more options, visit this group at http://groups.google.com/group/hazelcast?hl=en.
>
>

[Anatoly Polinsky]

Hi Talip,

Thanks for a quick reply.

Yes, I missed that, and will use a Map. The thing is though, I am
not after the actual data just yet, but for PKs themselves.

If PKs we iteratable ( e.g. 0,1,2,3,4, ... N ) I could just
calculate my partition size by count(*) / gridSize, and send [ min,
max ] PK pairs to each partition. For example: [0, 3] to partition 1,
[4, 7] to partition 2, etc.. But in this case I can't rely on PKs
being sequential OR iteratable, I only know that they are unique.

Hence I wanted to stage these PKs into a List ( well, now a
Map :) ) to give them a sequential access:
map.put( idGenerator.newId(), new CompositeKey( fieldOne,
fieldTwo ) ), and then I would know the map size, so I can follow the
original approach of sending [ min, max ] pairs to processing
partitions ( e.g. which can be threads, hazelcast nodes, gridgain
nodes, etc.. depending on how heavy the processing is )...

I am trying to find the most efficient way to deliver all these PKs
form DB to Hazelcast: e.g. cursor based data streaming, pagination,
single select, etc.. Have you ever had to stage something as big as
10*X millions of DB rows ( in this case each row is two fields =
composite key ) into a distributed map? If yes, how did you do that,
if no, what would you recommend?

Thank you,
/Anatoly

[Talip Ozturk]

>   If PKs we iteratable ( e.g. 0,1,2,3,4, ... N ) I could just
> calculate my partition size by count(*) / gridSize, and send [ min,
> max ] PK pairs to each partition. For example: [0, 3] to partition 1,
> [4, 7] to partition 2, etc.. But in this case I can't rely on PKs

This will help but here is the faster way doing it:
On each node read all the key fields (fieldOne, fieldTwo) from the database.
For each key:
Create CompositeKey(fieldOne, fieldTwo) = theKey
Check if theKey is locally owned.
Hazelcast.getPartitionService().getPartition(theKey).getOwner().localMember()
If locally owned then read the entire row and put it into the map.
map.put(new CompositeKey(fieldOne, fieldTwo), RowObject);

We have done it for millions of rows. The benefit is that every node
is putting the locally owned keys so the cost of the put is minimum.
Each node is participating in the loading process and each only doing
the local puts. You just have to make sure that there is no migration
during that process (listen for the migrations), if there is then loop
through all the keys one more time and check each exists.

-talip

[Anatoly Polinsky]

interesting, thanks for the recommendation.

I probably did not explain my self correctly, since the final goal is
to process the actual data ( RowObjects ) _later on_ by a pluggable
grid mechanism ( hence this can be just a grid out of threads,
hazelcast nodes, gridgain nodes, etc.. depending on the processing
weight and the record size ).

And I am using Hazelcast in a way as I would create and index region
on a table ( if I could ): just to have a sequential access to PKs. In
fact the system, this is being done for, won't probably have enough
resources to keep the actual data in memory at once. ( it would have
resources to keep PKs though ).

Once I have these Maps in form of [ unique and iteratable key,
Composite PK ], then later on, in the "second round of processing", I
have a luxury to calculate {min, max} pairs for each node ( not
necessarily Hazelcast node, just a "grid node" ) in order to partition
the processing.

I am looking for the most efficient way to get it all from DB to
Hazelcast. For example, if I do a single select for 70 millions
records, DB may become really upset, so I was thinking to maybe use
some kind of pagination ( based on ROW_NUMBER / RANK / etc.. ), or
maybe stream it through to Hazelcast map via DB cursor. I see that you
are offering to use the fact that I already have several Hazelcast
nodes, but I wanted to see if I can read those PKs in a _single_
location and put the into a _distributed_ map.

Thank you Talip,
/Anatoly