Cassandra with PHP - 在调用 cassandra-test.php 时，我得到“Call to undefined method CassandraClient::batch_insert()”

Question

我目前正在尝试让 Cassandra 在 Windows 7 上与 PHP 一起运行。 我安装了 cassandra 和 thrift... 当我调用 cassandra-test.php 时，出现以下错误：

( ! ) 致命错误：调用未定义的方法

CassandraClient::batch_insert() 在 C:\xampp\htdocs\YiiPlayground\cassandra-test.php 第 75 行 调用栈 # 时间记忆功能位置 1 0.0014 337552 {main}( ) ..\cassandra-test.php:0 2 0.0138 776232 CassandraDB->插入记录( ) ..\cassandra-test.php:304

cassandra-test.php 如下所示：

<?php
// CassandraDB version 0.1
// Software Projects Inc
// http://www.softwareprojects.com
//

// Includes
$GLOBALS['THRIFT_ROOT'] = 'C:/xampp/htdocs/Yii/kallaspriit-Cassandra-PHP-Client-Library/thrift';
//$GLOBALS['THRIFT_ROOT'] = realpath('E:/00-REGIESTART/Programme/Cassandra/thrift');
require_once $GLOBALS['THRIFT_ROOT'].'/packages/cassandra/Cassandra.php';
require_once $GLOBALS['THRIFT_ROOT'].'/packages/cassandra/cassandra_types.php';
require_once $GLOBALS['THRIFT_ROOT'].'/transport/TSocket.php';
require_once $GLOBALS['THRIFT_ROOT'].'/protocol/TBinaryProtocol.php';
require_once $GLOBALS['THRIFT_ROOT'].'/transport/TFramedTransport.php';
require_once $GLOBALS['THRIFT_ROOT'].'/transport/TBufferedTransport.php';

class CassandraDB
{
  // Internal variables
  protected $socket;
  protected $client;
  protected $keyspace;
  protected $transport;
  protected $protocol;
  protected $err_str = "";
  protected $display_errors = 0;
  protected $consistency = 1;
  protected $parse_columns = 1;

  // Functions

  // Constructor - Connect to Cassandra via Thrift
  function CassandraDB  ($keyspace, $host = "127.0.0.1", $port = 9160)
  {
    // Initialize
    $this->err_str = '';

    try
    {
    // Store passed 'keyspace' in object
    $this->keyspace = $keyspace;

    // Make a connection to the Thrift interface to Cassandra
    $this->socket = new TSocket($host, $port);
    $this->transport = new TFramedTransport($this->socket, 1024, 1024);
    $this->protocol = new TBinaryProtocolAccelerated($this->transport);
    $this->client = new CassandraClient($this->protocol);
    $this->transport->open();
    }
    catch (TException $tx)
    {
      // Error occured
      $this->err_str = $tx->why;
      $this->Debug($tx->why." ".$tx->getMessage());
    }
  }

  // Insert Column into ColumnFamily
  // (Equivalent to RDBMS Insert record to a table)
  function InsertRecord  ($table /* ColumnFamily */, $key /* ColumnFamily Key */, $record /* Columns */)
  {
    // Initialize
    $this->err_str = '';

    try
    {
      // Timestamp for update
      $timestamp = time();

      // Build batch mutation
      $cfmap = array();
      $cfmap[$table] = $this->array_to_supercolumns_or_columns($record, $timestamp);

      // Insert
      $this->client->batch_insert($this->keyspace, $key, $cfmap, $this->consistency);

      // If we're up to here, all is well
      $result = 1;
    }
    catch (TException $tx)
    {
      // Error occured
      $result = 0;
      $this->err_str = $tx->why;
      $this->Debug($tx->why." ".$tx->getMessage());
    }

    // Return result
    return $result;
  }

  // Insert SuperColumn into SuperColumnFamily
  // (Equivalent to RDMBS Insert record to a "nested table")
  function InsertRecordArray  ($table /* SuperColumnFamily */, $key_parent /* Super CF */,
                $record /* Columns */)
  {
    // Initialize
    $err_str = '';

    try
    {
      // Timestamp for update
      $timestamp = time();

      // Build batch mutation
      $cfmap = array();
      $cfmap[$table] = $this->array_to_supercolumns_or_columns($record, $timestamp);

      // Insert
      $this->client->batch_insert($this->keyspace, $key_parent, $cfmap, $this->consistency);

      // If we're up to here, all is well
      $result = 1;
    }
    catch (TException $tx)
    {
      // Error occured
      $result = 0;
      $this->err_str = $tx->why;
      $this->Debug($tx->why." ".$tx->getMessage());
    }

    // Return result
    return $result;
  }

  // Get record by key
  function GetRecordByKey  ($table /* ColumnFamily or SuperColumnFamily */, $key, $start_from="", $end_at="")
  {
    // Initialize
    $err_str = '';

    try
    {
      return $this->get($table, $key, NULL, $start_from, $end_at);
    }
    catch (TException $tx)
    {
      // Error occured
      $this->err_str = $tx->why;
      $this->Debug($tx->why." ".$tx->getMessage());
      return array();
    }
  }

  // Print debug message
  function Debug      ($str)
  {
    // If verbose is off, we're done
    if (!$this->display_errors) return;

    // Print
    echo date("Y-m-d h:i:s")." CassandraDB ERROR: $str\r\n";
  }

  // Turn verbose debug on/off (Default is off)
  function SetDisplayErrors($flag)
  {
    $this->display_errors = $flag;
  }

  // Set Consistency level (Default is 1)
  function SetConsistency  ($consistency)
  {
    $this->consistency = $consistency;
  }

  // Build cf array
  function array_to_supercolumns_or_columns($array, $timestamp=null)
  {
    if(empty($timestamp)) $timestamp = time();

    $ret = null;
    foreach($array as $name => $value) {
      $c_or_sc = new cassandra_ColumnOrSuperColumn();
      if(is_array($value)) {
        $c_or_sc->super_column = new cassandra_SuperColumn();
        $c_or_sc->super_column->name = $this->unparse_column_name($name, true);
        $c_or_sc->super_column->columns = $this->array_to_columns($value, $timestamp);
        $c_or_sc->super_column->timestamp = $timestamp;
      }
      else
      {
        $c_or_sc = new cassandra_ColumnOrSuperColumn();
        $c_or_sc->column = new cassandra_Column();
        $c_or_sc->column->name = $this->unparse_column_name($name, true);
        $c_or_sc->column->value = $value;
        $c_or_sc->column->timestamp = $timestamp;
      }
      $ret[] = $c_or_sc;
    }

    return $ret;
  }


  // Parse column names for Cassandra
  function parse_column_name($column_name, $is_column=true)
  {
    if(!$column_name) return NULL;

    return $column_name;
  }

  // Unparse column names for Cassandra
  function unparse_column_name($column_name, $is_column=true)
  {
    if(!$column_name) return NULL;

    return $column_name;
  }

  // Convert supercolumns or columns into an array
  function supercolumns_or_columns_to_array($array)
  {
    $ret = null;
    for ($i=0; $i<count($array); $i++)
    foreach ($array[$i] as $object)
    {
      if ($object)
      {
        // If supercolumn
        if (isset($object->columns))
        {
          $record = array();
          for ($j=0; $j<count($object->columns); $j++)
          {
            $column = $object->columns[$j];
            $record[$column->name] = $column->value;
          }
          $ret[$object->name] = $record;
        }
        // (Otherwise - not supercolumn)
        else
        {
          $ret[$object->name] = $object->value;
        }
      }
    }

    return $ret;
  }

  // Get record from Cassandra
  function get($table, $key, $super_column=NULL, $slice_start="", $slice_finish="")
  {
    try
    {
    $column_parent = new cassandra_ColumnParent();
    $column_parent->column_family = $table;
    $column_parent->super_column = $this->unparse_column_name($super_column, false);

    $slice_range = new cassandra_SliceRange();
    $slice_range->start = $slice_start;
    $slice_range->finish = $slice_finish;
    $predicate = new cassandra_SlicePredicate();
    $predicate->slice_range = $slice_range;

    $resp = $this->client->get_slice($this->keyspace, $key, $column_parent, $predicate, $this->consistency);

    return $this->supercolumns_or_columns_to_array($resp);
    }
    catch (TException $tx)
    {
    $this->Debug($tx->why." ".$tx->getMessage());
    return array();
    }
  }

  // Convert array to columns
  function array_to_columns($array, $timestamp=null) {
    if(empty($timestamp)) $timestamp = time();

    $ret = null;
    foreach($array as $name => $value) {
      $column = new cassandra_Column();
      $column->name = $this->unparse_column_name($name, false);
      $column->value = $value;
      $column->timestamp = $timestamp;

      $ret[] = $column;
    }
    return $ret;
  }

  // Get error string
  function ErrorStr()
  {
    return $this->err_str;
  }
} 


// Initialize Cassandra
$cassandra = new CassandraDB("SPI");

// Debug on
$cassandra->SetDisplayErrors(true);

// Insert record ("Columns" in Cassandra)
$record = array();
$record["name"] = "Mike Peters";
$record["email"] = "mike at softwareprojects.com";
if ($cassandra->InsertRecord('mytable', "Mike Peters", $record)) {
    echo "Record (Columns) inserted successfully.\r\n";
}

// Print record
$record = $cassandra->GetRecordByKey('mytable', "Mike Peters");
print_r($record);
?>

对此有什么想法，如何解决这个问题？ 多谢！

Answer 1

如果可以避免的话，你真的不想手工进行 Thrift。看一下 phpcassa 库：
https://github.com/thobbs/phpcassa

哦，在上面，看起来你想要 ' ln 上的“batch_mutate”而不是“batch_insert”。 75. 该方法在 cassandra 版本中更改了名称 > 0.6.x