小编Ste*_*vva的帖子

我正在使用这些参数在DynamoDB中创建一个表和GSI,根据文档:

configId是表的主键,我使用它publisherId作为GSI的主键.(为简洁起见,我删除了一些不必要的配置参数)

var params = {
    TableName: 'Configs',
    KeySchema: [ 
        {
            AttributeName: 'configId',
            KeyType: 'HASH',
        }
    ],
    AttributeDefinitions: [
        {
            AttributeName: 'configId',
            AttributeType: 'S',
        },
        {
            AttributeName: 'publisherId',
            AttributeType: 'S',
        }
    ],
    GlobalSecondaryIndexes: [ 
        { 
            IndexName: 'publisher_index', 
            KeySchema: [
                {
                    AttributeName: 'publisherId',
                    KeyType: 'HASH',
                }
            ]
        }
    ]
};

我用这个查询这个表:

{ TableName: 'Configs',
  IndexName: 'publisher_index',
  KeyConditionExpression: 'publisherId = :pub_id',
  ExpressionAttributeValues: { ':pub_id': { S: '700' } } }

但我一直收到错误:

"ValidationException:一个或多个参数值无效:条件参数类型与模式类型不匹配"

在文档中,它指定主要KeyType可以是HASH或RANGE,并且您 …

我正在尝试使用我将用于构建实时数据管道的技术,并且我遇到了将我的内容导出到文件的一些问题.

我已经设置了一个本地kafka集群和一个node.js生成器,它只发送一条简单的文本消息来测试功能并粗略估计实现的复杂性.

这是从kafka读取的火花流工作,我试图让它写入文件.

from pyspark import SparkContext
from pyspark.streaming import StreamingContext
from pyspark.streaming.kafka import KafkaUtils

# Create a local StreamingContext with two working thread and batch interval of 1 second
sc = SparkContext("local[2]", "KafkaStreamingConsumer")
ssc = StreamingContext(sc, 10)

kafkaStream = KafkaUtils.createStream(ssc, "localhost:2181", "consumer-group", {"test": 1})

kafkaStream.saveAsTextFile('out.txt')

print 'Event recieved in window: ', kafkaStream.pprint()

ssc.start()
ssc.awaitTermination()

我在提交spark工作时看到的错误是:

kafkaStream.saveAsTextFile('out.txt')
AttributeError: 'TransformedDStream' object has no attribute 'saveAsTextFile'

没有对数据执行计算或转换,我只想构建流程.我需要更改/添加什么才能导出文件中的数据？

当我转换的字符串是int时,我似乎无法弄清楚为什么它会抛出NFE

[码]

 public void setCurrentTransferRate(){
    try{
        long startTime = System.currentTimeMillis();  
        String[] command = {"/bin/sh", "-c", "ifconfig " + interface_name + " | grep -oP     'RX bytes:[0-9]{1,11}'"};
        String[] command1 = {"/bin/sh", "-c", "ifconfig " + interface_name + " | grep -oP 'TX bytes:[0-9]{1,11}'"};
        Process child = Runtime.getRuntime().exec(command);
        Process child1 = Runtime.getRuntime().exec(command1);
        BufferedReader r = new BufferedReader(new InputStreamReader(child.getInputStream()));       //i prwti metrisi twn RX kai TX bytes
        BufferedReader r1 = new BufferedReader(new InputStreamReader(child1.getInputStream()));
        String temp = r.readLine();
        temp = temp.replace("RX bytes:","");   
        String temp1 …

这个命令给我以下输出:

 cat /proc/net/dev | awk '{print $1}'
 Inter-|
 face
 eth0:
 lo:
 wlan0:

有没有办法解雇线间,面,所以我只能得到接口的名称？