COPY cassandra table from csv file

Question

Я создаю демонстрационный ландшафт для Cassandra, Apache Spark и Flume на моем Mac (Mac OS X Yosemite с Oracle jdk1.7.0_55). Пейзаж должен служить доказательством концепции новой платформы аналитики, поэтому мне нужны некоторые тестовые данные в моей кассандре db. Я использую cassandra 2.0.8.

Я создал некоторые демо-данные в excel и экспортировал их как CSV-файл. Структура такова:

ProcessUUID;ProcessID;ProcessNumber;ProcessName;ProcessStartTime;ProcessStartTimeUUID;ProcessEndTime;ProcessEndTimeUUID;ProcessStatus;Orderer;VorgangsNummer;VehicleID;FIN;Reference;ReferenceType 
0F0D1498-D149-4FCC-87C9-F12783FDF769;AbmeldungKl‰rfall;1;Abmeldung Kl‰rfall;2011-02-03 04:05+0000;;2011-02-17 04:05+0000;;Finished;SIXT;4278;A-XA 1;WAU2345CX67890876;KLA-BR4278;internal 

Затем я создал пространство ключей и семейство столбцов в cqlsh с помощью:

CREATE KEYSPACE dadcargate 
WITH REPLICATAION = { 'class' : 'SimpleStrategy', 'replication_factor' : '1' }; 

use dadcargate; 

CREATE COLUMNFAMILY Process (
    ProcessUUID uuid, ProcessID varchar, ProcessNumber bigint, ProcessName varchar, 
    ProcessStartTime timestamp, ProcessStartTimeUUID timeuuid, ProcessEndTime timestamp, 
    ProcessEndTimeUUID timeuuid, ProcessStatus varchar, Orderer varchar, 
    VorgangsNummer varchar, VehicleID varchar, FIN varchar, Reference varchar, 
    ReferenceType varchar, 
PRIMARY KEY (ProcessUUID)) 
WITH COMMENT='A process is like a bracket around multiple process steps'; 

имя столбца семьи и все столбцы в нем создаются все строчными буквами - будет должны также расследовать это, но это не так актуально в настоящий момент.

Теперь я беру мой файл CSV, который имеет около 1600 записей и хочу импортировать, что в моей таблице с именем process так:

cqlsh:dadcargate> COPY process (processuuid, processid, processnumber, processname, 
processstarttime, processendtime, processstatus, orderer, vorgangsnummer, vehicleid, 
fin, reference, referencetype) 
FROM 'Process_BulkData.csv' WITH DELIMITER = ';' AND HEADER = TRUE; 

Это дает следующее сообщение об ошибке:

Record #0 (line 1) has the wrong number of fields (15 instead of 13). 
0 rows imported in 0.050 seconds. 

Каких по сути, истинно, поскольку у меня нет поля timeUUID в моем cvs-export.

Если я пытаюсь команду COPY без явных столбцов-имен, как это (учитывая тот факт, что я на самом деле пропустить два поля):

cqlsh:dadcargate> COPY process from 'Process_BulkData.csv' 
WITH DELIMITER = ';' AND HEADER = TRUE; 

Я в конечном итоге с другой ошибкой:

Bad Request: Input length = 1 
Aborting import at record #0 (line 1). Previously-inserted values still present. 
0 rows imported in 0.009 seconds. 

Hm. Это странно, но все в порядке. Возможно, команде COPY не нравится тот факт, что отсутствуют два поля. Я все еще считаю это странным, поскольку недостающие поля, конечно, там (с структурной точки зрения), но только пустые.

Я до сих пор есть еще один выстрел: Я удалил недостающие столбцы в Excel, снова экспортировал файл как CVS и попытаться импортировать без строки заголовка в моем CSV НО явных имен столбцов, как это:

cqlsh:dadcargate> COPY process (processuuid, processid, processnumber, processname, 
processstarttime, processendtime, processstatus, orderer, vorgangsnummer, vehicleid, 
fin, reference, referencetype) 
FROM 'Process_BulkData-2.csv' WITH DELIMITER = ';' AND HEADER = TRUE; 

I получить эту ошибку:

Bad Request: Input length = 1 
Aborting import at record #0 (line 1). Previously-inserted values still present. 
0 rows imported in 0.034 seconds. 

Может ли кто-нибудь сказать мне, что я здесь делаю неправильно? Согласно documentation of copy-command, способ, которым я настраиваю свои команды, должен работать как минимум для двух из них. Или я бы подумал.

Но, я, очевидно, пропустил что-то важное.

Answer 1

Команда cqlsh's COPY может быть обидчивой.Однако в COPY documentation эта линия:

The number of columns in the CSV input is the same as the number of columns in the Cassandra table metadata.

Имея это в-виду, мне удалось получить ваши данные для импорта с COPY FROM, называя пустые поля (processstarttimeuuid и processendtimeuuid соответственно):

aploetz@cqlsh:stackoverflow> COPY process (processuuid, processid, processnumber, 
processname, processstarttime, processstarttimeuuid, processendtime, 
processendtimeuuid, processstatus, orderer, vorgangsnummer, vehicleid, fin, reference, 
referencetype) FROM 'Process_BulkData.csv' WITH DELIMITER = ';' AND HEADER = TRUE; 

1 rows imported in 0.018 seconds. 
aploetz@cqlsh:stackoverflow> SELECT * FROM process ; 

processuuid       | fin    | orderer | processendtime   | processendtimeuuid | processid   | processname  | processnumber | processstarttime   | processstarttimeuuid | processstatus | reference | referencetype | vehicleid | vorgangsnummer 
--------------------------------------+-------------------+---------+---------------------------+--------------------+-------------------+--------------------+---------------+---------------------------+----------------------+---------------+------------+---------------+-----------+---------------- 
0f0d1498-d149-4fcc-87c9-f12783fdf769 | WAU2345CX67890876 | SIXT | 2011-02-16 22:05:00+-0600 |    null | AbmeldungKl‰rfall | Abmeldung Kl‰rfall |    1 | 2011-02-02 22:05:00+-0600 |     null |  Finished | KLA-BR4278 |  internal | A-XA 1 |   4278 

(1 rows) 

Answer 2

Загрузка файла CSV в Кассандре таблице

step1) установить Кассандры погрузчик с помощью этого URL
SUDO Wget https://github.com/brianmhess/cassandra-loader/releases/download/v0.0.23/cassandra-loader

step2) Sudo CHMOD + х Cassandra-погрузчик

а) имя файла CSV "pt_bms_tkt_success_record_details_new_2016_12_082017-01-0312-30-01.csv"

б) Название это пространство ключей "bms_test"

с) название таблицы является "pt_bms_tkt_success_record_details_new"

г) столбцы "trx_id ...... trx_day"

step3) CSV файл Locati на и Cassandra-погрузчик "cassandra3.7/бен /"

шаг $) [СТП @ RIL-SRV-sp3 бин] $ ./cassandra-loader -f pt_bms_tkt_success_record_details_new_2016_12_082017-01-0312-30-01.csv -host 192.168.1.29 -schema «bms_test.pt_bms_tkt_success_record_details_new (trx_id, max_seq, trx_type, trx_record_type, trx_date, trx_show_date, cinema_str_id, session_id, ttype_code, item_id, item_var_sequence, trx_booking_id, venue_name, screen_by_tnum, price_group_code, area_cat_str_code, area_by_tnum, venue_capacity, amount_currentprice, venue_class, trx_booking_status_committed, booking_status, amount_paymentstatus, event_application, venue_cinema_companyname, venue_cinema_name, venue_cinema_type, venue_cinema_application, region_str_code, venue_city_name, sub_region_str_code, sub_region_str_name, EVENT_CODE, EVENT_TYPE, event_name, event_language, event_genre, event_censor_rating, event_release_date, event_producer_code, event_item_na меня, event_itemvariable_name, event_quantity, amount_amount, amount_bookingfee, amount_deliveryfee, amount_additionalcharges, amount_final, amount_tax, offer_isapplied, offer_type, OFFER_NAME, offer_amount, payment_lastmode, payment_lastamount, payment_reference1, payment_reference2, payment_bank, customer_loginid, customer_loginstring, offer_referral, customer_mailid, customer_mobile, trans_str_sales_status_at_venue, trans_mny_trans_value_at_venue, payment_ismypayment, click_recordsource, кампания, источник, ключевое слово, средняя, ​​venue_multiplex, venue_state, mobile_type, transaction_range, life_cyclestate_from, transactions_after_offer, is_premium_transaction, city_type, holiday_season, week_type, event_popularity, transactionrange_after_discount, showminusbooking, input_source_name, канал, time_stamp, life_cyclestate_to, record_status, week_name, number_of_active_customers, event_genre1, event_genre2, event_genre3, event_genre4, event_language1, event_language2, event_language3, event_language4, event_release_date_range, showminusbooking_range, Рез erve1, резерв 2, reserve3, reserve4, reserve5, payment_mode, payment_type, date_of_first_transaction, transaction_time_in_hours, showtime_in_hours, trx_day) ";