我正在将多行记录文本文件读入 RDD。底层数据是这样的

Time    MHist::852-YF-007   
2016-05-10 00:00:00 0
2016-05-09 23:59:00 0
2016-05-09 23:58:00 0
Time    MHist::852-YF-008   
2016-05-10 00:00:00 0
2016-05-09 23:59:00 0
2016-05-09 23:58:00 0

不，我想转换 RDD，以便获得键映射(时间戳、值)。这可以分几个步骤完成。但我只想在一次调用中提取该信息(但在 Python 2.7 中不是 3)。

RDD是这样的:

[(0, u''),
 (12,
  u'852-YF-007\t\r\n2016-05-10 00:00:00\t0\r\n2016-05-09 23:59:00\t0\r\n2016-05-09 23:58:00\t0\r\n2016-05-09 23:57:00\t0\r\n2016-05-09 23:56:00\t0\r\n2016-05-09 23:55:00\t0\r\n2016-05-09 23:54:00\t0\r\n2016-05-09 23:53:00\t0\r\n2016-05-09 23:52:00\t0\r\n2016-05-09 23:51:00\t0\r\n2016-05-09 23:50:00\t0\r\n2016-05-09 23:49:00\t0\r\n2016-05-09 23:48:00\t0\r\n2016-05-09 23:47:00\t0\r\n2016-05-09 23:46:00\t0\r\n2016-05-09 23:45:00\t0\r\n2016-05-09 23:44:00\t0\r\n2016-05-09 23:43:00\t0\r\n2016-05-09 23:42:00\t0\n'),
 (473,
  u'852-YF-008\t\r\n2016-05-10 00:00:00\t0\r\n2016-05-09 23:59:00\t0\r\n2016-05-09 23:58:00\t0\r\n2016-05-09 23:57:00\t0\r\n2016-05-09 23:56:00\t0\r\n2016-05-09 23:55:00\t0\r\n2016-05-09 23:54:00\t0\r\n2016-05-09 23:53:00\t0\r\n2016-05-09 23:52:00\t0\r\n2016-05-09 23:51:00\t0\r\n2016-05-09 23:50:00\t0\r\n2016-05-09 23:49:00\t0\r\n2016-05-09 23:48:00\t0\r\n2016-05-09 23:47:00\t0\r\n2016-05-09 23:46:00\t0\r\n2016-05-09 23:45:00\t0\r\n2016-05-09 23:44:00\t0\r\n2016-05-09 23:43:00\t0\r\n2016-05-09 23:42:00\t0')]

对于每一对，有趣的部分是值(内容)。在该值中，第一项是键/名称，其余是带有时间戳的值。因此，我正在尝试使用它:

sheet = sc.newAPIHadoopFile(
    'sample.txt',
    'org.apache.hadoop.mapreduce.lib.input.TextInputFormat',
    'org.apache.hadoop.io.LongWritable',
    'org.apache.hadoop.io.Text',
    conf={'textinputformat.record.delimiter': 'Time\tMHist::'}
)

from operator import itemgetter

def process(pair):
    _, content = pair
    if not content: 
        pass

    lines = content.splitlines();
    #k = lines[0].strip()
    #vs =lines[1:]
    k, vs = itemgetter(0, slice(1, None), lines)
    #k, *vs = [x.strip() for x in content.splitlines()]  # Python 3 syntax

    for v in vs:
        try:
            ds, x = v.split("\t")
            yield k, (dateutil.parser.parse(ds), float(x))  # or int(x)
            return
        except ValueError:
            pass

sheet.flatMap(process).take(5)

但是我得到这个错误:

TypeError: 'operator.itemgetter' object is not iterable

进入函数的对具有字符位置(我可以忽略)和内容。内容应按\r\n 拆分，线数组的第一项是键，而其他项则作为 flatMap 的键-时间戳-值。

那么，我的流程方法哪里做错了？

同时，由于 Stackoverflow 和所有其他人的帮助，我想出了这个解决方案。这个工作得很好:

# reads a text file in TSV notation having the key-value no as first column but 
# as a randomly occuring line followed by its values. Remark: a variable might occur in several files

#Time    MHist::852-YF-007   
#2016-05-10 00:00:00 0
#2016-05-09 23:59:00 0
#2016-05-09 23:58:00 0
#Time    MHist::852-YF-008   
#2016-05-10 00:00:00 0
#2016-05-09 23:59:00 0
#2016-05-09 23:58:00 0

#imports
from operator import itemgetter
from datetime import datetime

#read the text file with special record-delimiter --> all lines after Time\tMHist:: are the values for that variable
sheet = sc.newAPIHadoopFile(
    'sample.txt',
    'org.apache.hadoop.mapreduce.lib.input.TextInputFormat',
    'org.apache.hadoop.io.LongWritable',
    'org.apache.hadoop.io.Text',
    conf={'textinputformat.record.delimiter': 'Time\tMHist::'}
)

#this avoid using multiple map/flatMap/mapValues/flatMapValues calls by extracting the values at once
def process_and_extract(pair):
    # first part will be the char-position within the file, which we can ignore
    # second is the real content as one string and not yet splitted
    _, content = pair
    if not content: 
        pass

    try:
        # once the content is split into lines:
        # 1. the first line will have the bare variable name since we removed the preceeding 
        # part when opening the file (see delimiter above)
        # 2. the second line until the end will include the values for the current variable

        # Python 2.7 syntax
        #clean = itemgetter(0, slice(1, None))(lines)
        clean = [x.strip() for x in content.splitlines()]
        k, vs = clean[0], clean[1:]    

        # Python 3 syntax
        #k, *vs = [x.strip() for x in content.splitlines()] 
        #for v in vs*:

        for v in vs:
            try:
                # split timestamp and value and convert (cast) them from string to correct data type
                ds, x = v.split("\t")
                yield k, (datetime.strptime(ds, "%Y-%m-%d %H:%M:%S"), float(x))
            except ValueError:
                # might occur if a line format is corrupt
                pass
    except IndexError:
        # might occur if content is empty or iregular
        pass

# read, flatten, extract and reduce the file at once        
sheet.flatMap(process_and_extract) \
    .reduceByKey(lambda x, y: x + y) \
    .take(5)

第二个版本避免了 for-each-loop，最终快了 20%:

start_time = time.time()

#read the text file with special record-delimiter --> all lines after Time\tMHist:: are the values for that variable
sheet = sc.newAPIHadoopFile(
    'sample.txt',
    'org.apache.hadoop.mapreduce.lib.input.TextInputFormat',
    'org.apache.hadoop.io.LongWritable',
    'org.apache.hadoop.io.Text',
    conf={'textinputformat.record.delimiter': 'Time\tMHist::'}
)

def extract_blob(pair):
    if not pair:
        pass

    try:
        offset, content = pair
        if not content: 
            pass

        clean = [x.strip() for x in content.splitlines()]
        if not clean or len(clean) < 2:
            pass

        k, vs = clean[0], clean[1:]
        if not k:
            pass

        return k.strip(), vs
    except IndexError:
        # might occur if content is empty or malformed
        pass

def extract_line(pair):
    if not pair:
        pass

    key, line = pair;
    if not key or not line:
        pass

    # split timestamp and value and convert (cast) them from string to correct data type
    content = line.split("\t")
    if not content or len(content) < 2:
        pass

    try:
        ds, x = content
        if not ds or not x:
            pass 

        return (key, datetime.strptime(ds, "%Y-%m-%d %H:%M:%S"), float(x))
    except ValueError:
        # might occur if a line format is corrupt
        pass

def check_empty(x):
    return not (x == None)

#drop keys and filter out non-empty entries
non_empty = sheet.filter(lambda (k, v): v)

#group lines having variable name at first line
grouped_lines = non_empty.map(extract_blob)

#extract variable name and split it from the variable values
flat_lines = grouped_lines.flatMapValues(lambda x: x)

#extract the values from the value
flat_triples = flat_lines.map(extract_line).filter(check_empty)

#convert to dataframe
df = flat_triples.toDF(["Variable", "Time", "Value"])

df.write \
    .partitionBy("Variable") \
    .saveAsTable('Observations', format='parquet', mode='overwrite', path=output_hdfs_filepath)

print("loading and saving done in {} seconds".format(time.time() - start_time));

最佳答案

itemgetter 返回一个函数，该函数接受一个对象并为传递给 itemgetter 的每个参数调用 __getitem__。所以你必须在 lines 上调用它:

itemgetter(0, slice(1, None))(lines)

大致相当于

[lines[i] for i in [0, slice(1, None)])

lines[slice(1, None)] 基本上是 lines[1:]。

这意味着您必须首先确保 lines 不为空，否则 lines[0] 将失败。

if lines:  # bool(empty_sequence) is False
    k, vs = itemgetter(0, slice(1, None))(lines)
    for v in vs:
        ...

将它们放在一起，包括 doctests:

def process(pair):
    r"""
    >>> list(process((0, u'')))
    []
    >>> kvs = list(process((
    ... 12,
    ... u'852-YF-007\t\r\n2016-05-10 00:00:00\t0\r\n2016-05-09 23:59:00\t0')))
    >>> kvs[0] 
    (u'852-YF-007', (datetime.datetime(2016, 5, 10, 0, 0), 0.0))
    >>> kvs[1]
    (u'852-YF-007', (datetime.datetime(2016, 5, 9, 23, 59), 0.0))
    >>> list(process((
    ... 10,
    ... u'852-YF-007\t\r\n2ad-05-10 00')))
    []
    """ 
    _, content = pair
    clean = [x.strip() for x in content.strip().splitlines()]

    if clean:
        k, vs = itemgetter(0, slice(1, None))(clean)
        for v in vs:
            try:
                ds, x = v.split("\t")
                yield k, (datetime.strptime(ds, "%Y-%m-%d %H:%M:%S"), float(x))
            except ValueError:
                pass 

关于python - 如何在 flatMap 函数中实现迭代，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/38126919/