ebay / cassandra-river Goto Github PK

We have a cassandra cluster, and want to index all data from differetens CF's to an elasticsearch.

We want to use cassandra-river but we have a problem when building this

at "mvn clean package" when Test starts, we get this errors.

Exception in thread "Queue-Indexer-thread-408" java.lang.NullPointerException
at org.elasticsearch.river.cassandra.CassandraRiver$Indexer.run(CassandraRiver.java:149)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:744)

Any help would be appreciated!

I have successfully installed the river and can successfully connect to the cassandra cluster, however I am missing something fundamental since I cannot see a index building.

I have 10,000 rows in my CF in a 3 node cluster
and see activity in the ES logs
[2014-02-27 16:01:03,516][INFO ][river.cassandra ] [Warren III Worthington] [cassandra][maccasIndex] Starting thread with 1000 keys
[2014-02-27 16:01:03,716][INFO ][river.cassandra ] [Warren III Worthington] [cassandra][maccasIndex] Starting thread with 738 keys

this repeats over and over again.

I am not using Auth in the cluster so have omitted the properties

sanitised

curl -XPUT 'localhost:9200/_river/maccasIndex/_meta' -d '{
"type" : "cassandra",
"cassandra" : {
"cluster_name" : "MaccasCluster",
"keyspace" : "MaccasKeyspace",
"column_family" : "MaccasCF",
"batch_size" : 1000,
"hosts" : "dsc1n1:9160,dsc1n2:9160,dsc1n3:9160"

},
"index" : {
    "index" : "maccasIndex",
    "type" : "cassandra"
}

}'

Cassandra info:
[cqlsh 3.1.2 | Cassandra 1.2.6.1 | CQL spec 3.0.0 | Thrift protocol 19.36.0]

the CF definition:

CREATE TABLE "MaccasCF" (
key text PRIMARY KEY,
"___class" text,
"addressLine1" text,
"addressLine2" text,
......

I am definitely missing something

http://localhost:9200/_status?pretty=1

{
"_shards" : {
"total" : 2,
"successful" : 1,
"failed" : 0
},
"indices" : {
"_river" : {
"index" : {
"primary_size_in_bytes" : 3820,
"size_in_bytes" : 3820
},
"translog" : {
"operations" : 0
},
"docs" : {
"num_docs" : 1,
"max_doc" : 1,
"deleted_docs" : 0
},
"merges" : {
"current" : 0,
"current_docs" : 0,
"current_size_in_bytes" : 0,
"total" : 0,
"total_time_in_millis" : 0,
"total_docs" : 0,
"total_size_in_bytes" : 0
},
"refresh" : {
"total" : 2,
"total_time_in_millis" : 84
},
"flush" : {
"total" : 1,
"total_time_in_millis" : 20
},
"shards" : {
"0" : [ {
"routing" : {
"state" : "STARTED",
"primary" : true,
"node" : "czJqPzFBSseiXoKX0YtWkA",
"relocating_node" : null,
"shard" : 0,
"index" : "_river"
},
"state" : "STARTED",
"index" : {
"size_in_bytes" : 3820
},
"translog" : {
"id" : 1393509195769,
"operations" : 0
},
"docs" : {
"num_docs" : 1,
"max_doc" : 1,
"deleted_docs" : 0
},
"merges" : {
"current" : 0,
"current_docs" : 0,
"current_size_in_bytes" : 0,
"total" : 0,
"total_time_in_millis" : 0,
"total_docs" : 0,
"total_size_in_bytes" : 0
},
"refresh" : {
"total" : 2,
"total_time_in_millis" : 84
},
"flush" : {
"total" : 1,
"total_time_in_millis" : 20
}
} ]
}
}
}
}

http://localhost:9200/_river/maccasIndex/_mapping?pretty=1
{
"_river" : {
"mappings" : {
"maccasIndex" : {
"properties" : {
"cassandra" : {
"properties" : {
"batch_size" : {
"type" : "long"
},
"cluster_name" : {
"type" : "string"
},
"column_family" : {
"type" : "string"
},
"hosts" : {
"type" : "string"
},
"keyspace" : {
"type" : "string"
}
}
},
"index" : {
"properties" : {
"index" : {
"type" : "string"
},
"type" : {
"type" : "string"
}
}
},
"type" : {
"type" : "string"
}
}
}
}
}

}

if I run:

http://localhost:9200/_river/maccasIndex/_search?q=_:_pretty=1

I get
{"took":268,"timed_out":false,"_shards":{"total":1,"successful":1,"failed":0},"hits":{"total":0,"max_score":null,"hits":[]}}

but no sign of any indexed data

Can you tell me where I am going wrong?

Hi,
Is there a way to tweak the data before indexing. e.g change field names or skip few fields?

Regards
Saud.