GitHubじゃ!Pythonじゃ!

GitHubからPython関係の優良リポジトリを探したかったのじゃー、でも英語は出来ないから日本語で読むのじゃー、英語社会世知辛いのじゃー

EntilZha

PyFunctional – チェーン関数プログラミングによるデータパイプラインを作成するためのPythonライブラリ

投稿日:

チェーン関数プログラミングによるデータパイプラインを作成するためのPythonライブラリ http://www.pyfunctional.org

PyFunctional

特徴

PyFunctionalは、連鎖した関数演算子を使用してデータパイプラインを簡単に作成します。 それができることのいくつかの例があります:

  • 連鎖演算子: seq(1, 2, 3).map(lambda x: x * 2).reduce(lambda x, y: x + y)
  • 表現力豊かなAPI
  • textcsvjsonjsonlsqlitegzipbz2 、およびlzma/xzファイルの読み書き
  • mapような「恥ずかしいほどパラレルな」操作を簡単に並列化する
  • 完全なドキュメント、厳密なユニットテストスイート、100%テストカバレッジ、および堅牢性を提供するCI

PyFunctionalのAPIは、Scalaコレクション、Apache Spark RDD、Microsoft LINQからインスピレーションを得ています。

目次

  1. インストール
    1. 簡単な例
    2. 集約と結合
    3. SQLite3の読み書き
    4. パンダとのデータ交換
  2. ファイルへの書き込み
  3. パラレル実行
  4. Github Shortformのドキュメント
    1. ストリーム、変換、およびアクション
    2. ストリームAPI
    3. 変換とアクションAPI
    4. 遅延実行
  5. 貢献とバグ修正
  6. 変更ログ

インストール

PyFunctionalpypiで利用可能で、 次のコマンドを実行することでインストールできます。

# Install from command line
$ pip install pyfunctional

その後、Pythonランで: from functional import seq

PyFunctionalは多くのタスクに便利で、いくつかの一般的なファイルタイプをネイティブに開くことができます。 あなたができることのいくつかの例を以下に示します。

簡単な例

from functional import seq

seq(1, 2, 3, 4)\
    .map(lambda x: x * 2)\
    .filter(lambda x: x > 4)\
    .reduce(lambda x, y: x + y)
# 14

# or if you don't like backslash continuation
(seq(1, 2, 3, 4)
    .map(lambda x: x * 2)
    .filter(lambda x: x > 4)
    .reduce(lambda x, y: x + y)
)
# 14

ストリーム、変換、およびアクション

PyFunctionalは3種類の関数があります:

  1. ストリーム:コレクションAPIで使用するためにデータを読み取ります。
  2. 変換: mapflat_mapfilterなどの関数を使用してストリームからデータを変換する
  3. アクション:これにより、一連の変換が具体的な値に評価されます。 to_listreduceto_dictはアクションの例です。

seq(1, 2, 3).map(lambda x: x * 2).reduce(lambda x, y: x + y)では、 seqはストリーム、 mapは変換、 reduceはアクションです。

アカウント取引のリストをフィルタリングする

from functional import seq
from collections import namedtuple

Transaction = namedtuple('Transaction', 'reason amount')
transactions = [
    Transaction('github', 7),
    Transaction('food', 10),
    Transaction('coffee', 5),
    Transaction('digitalocean', 5),
    Transaction('food', 5),
    Transaction('riotgames', 25),
    Transaction('food', 10),
    Transaction('amazon', 200),
    Transaction('paycheck', -1000)
]

# Using the Scala/Spark inspired APIs
food_cost = seq(transactions)\
    .filter(lambda x: x.reason == 'food')\
    .map(lambda x: x.amount).sum()

# Using the LINQ inspired APIs
food_cost = seq(transactions)\
    .where(lambda x: x.reason == 'food')\
    .select(lambda x: x.amount).sum()

# Using PyFunctional with fn
from fn import _
food_cost = seq(transactions).filter(_.reason == 'food').map(_.amount).sum()

集約と結合

アカウントトランザクションの例は、リスト内包表記を使って純粋なPythonで簡単に行うことができます。 PyFunctional優れていることのいくつかを示すには、いくつかの単語数の例を見てください。

words = 'I dont want to believe I want to know'.split(' ')
seq(words).map(lambda word: (word, 1)).reduce_by_key(lambda x, y: x + y)
# [('dont', 1), ('I', 2), ('to', 2), ('know', 1), ('want', 2), ('believe', 1)]

次の例では、メッセージとメタデータを含むjsonの行(jsonl)でフォーマットされたチャットログがあります。 典型的なjsonlファイルは、ファイルの各行に1つの有効なjsonを持ちます。 以下はexamples/chat_logs.jsonlからのいくつかの行examples/chat_logs.jsonl

{"message":"hello anyone there?","date":"10/09","user":"bob"}
{"message":"need some help with a program","date":"10/09","user":"bob"}
{"message":"sure thing. What do you need help with?","date":"10/09","user":"dave"}
from operator import add
import re
messages = seq.jsonl('examples/chat_logs.jsonl')

# Split words on space and normalize before doing word count
def extract_words(message):
    return re.sub('[^0-9a-z ]+', '', message.lower()).split(' ')


word_counts = messages\
    .map(lambda log: extract_words(log['message']))\
    .flatten().map(lambda word: (word, 1))\
    .reduce_by_key(add).order_by(lambda x: x[1])

次に、その例を続けるが、 examples/users.jsonからのユーザーのjsonデータベースを紹介する。 前の例では、 PyFunctionalが単語カウントを行う方法を示しました。次の例では、 PyFunctionalがさまざまなデータソースにどのように参加できるかを示しています。

# First read the json file
users = seq.json('examples/users.json')
#[('sarah',{'date_created':'08/08','news_email':True,'email':'sarah@gmail.com'}),...]

email_domains = users.map(lambda u: u[1]['email'].split('@')[1]).distinct()
# ['yahoo.com', 'python.org', 'gmail.com']

# Join users with their messages
message_tuples = messages.group_by(lambda m: m['user'])
data = users.inner_join(message_tuples)
# [('sarah',
#    (
#      {'date_created':'08/08','news_email':True,'email':'sarah@gmail.com'},
#      [{'date':'10/10','message':'what is a...','user':'sarah'}...]
#    )
#  ),...]

# From here you can imagine doing more complex analysis

CSV、集計関数、および集合関数

examples/camping_purchases.csvにはキャンプ用購入のリストがあります。 いくつかのコスト分析を行い、 examples/gear_list.txt格納されている必要なキャンプギアリストと比較します。

purchases = seq.csv('examples/camping_purchases.csv')
total_cost = purchases.select(lambda row: int(row[2])).sum()
# 1275

most_expensive_item = purchases.max_by(lambda row: int(row[2]))
# ['4', 'sleeping bag', ' 350']

purchased_list = purchases.select(lambda row: row[1])
gear_list = seq.open('examples/gear_list.txt').map(lambda row: row.strip())
missing_gear = gear_list.difference(purchased_list)
# ['water bottle','gas','toilet paper','lighter','spoons','sleeping pad',...]

上に示した集合関数( summax_by )に加えて、さらに多くの関数があります。 同様に、 differenceに加えていくつかのより類似した機能があります。

SQLite3の読み書き

PyFunctionalはSQLite3データベースファイルを読み書きできます。 次の例では、usersはusers examples/users.dbから読み込まれ、列はid:Intおよびname:String列として格納されます。

db_path = 'examples/users.db'
users = seq.sqlite3(db_path, 'select * from user').to_list()
# [(1, 'Tom'), (2, 'Jack'), (3, 'Jane'), (4, 'Stephan')]]

sorted_users = seq.sqlite3(db_path, 'select * from user order by name').to_list()
# [(2, 'Jack'), (3, 'Jane'), (4, 'Stephan'), (1, 'Tom')]

SQLite3データベースへの書き込みも同様に簡単です

import sqlite3
from collections import namedtuple

with sqlite3.connect(':memory:') as conn:
    conn.execute('CREATE TABLE user (id INT, name TEXT)')
    conn.commit()
    User = namedtuple('User', 'id name')
    
    # Write using a specific query
    seq([(1, 'pedro'), (2, 'fritz')]).to_sqlite3(conn, 'INSERT INTO user (id, name) VALUES (?, ?)')
    
    # Write by inserting values positionally from a tuple/list into named table
    seq([(3, 'sam'), (4, 'stan')]).to_sqlite3(conn, 'user')
    
    # Write by inferring schema from namedtuple
    seq([User(name='tom', id=5), User(name='keiga', id=6)]).to_sqlite3(conn, 'user')
    
    # Write by inferring schema from dict
    seq([dict(name='david', id=7), dict(name='jordan', id=8)]).to_sqlite3(conn, 'user')
    
    # Read everything back to make sure it wrote correctly
    print(list(conn.execute('SELECT * FROM user')))
    
    # [(1, 'pedro'), (2, 'fritz'), (3, 'sam'), (4, 'stan'), (5, 'tom'), (6, 'keiga'), (7, 'david'), (8, 'jordan')]

ファイルへの書き込み

PyFunctionalcsvjsonjsonlsqlite3 、およびテキストファイルから読み込むことができるのと同様に、それらを書き込むこともできます。 完全なAPIドキュメントについては、コレクションAPIテーブルまたは公式ドキュメントを参照してください。

圧縮ファイル

PyFunctionalは、 gziplzma/xzbz2圧縮されたファイルを自動的に検出します。 これは、圧縮されているかどうかを判断するためにファイルの最初の数バイトを調べることによって行われるため、コードを変更する必要はありません。

圧縮されたファイルを書き込むために、すべてのto_関数には、 compression Noneの場合はデフォルトNoneに、 gzip圧縮の場合はgzipまたはgzlzma圧縮の場合はlzmaまたはxz 、bz2圧縮の場合はbz2に設定できます。

パラレル実行

並列性を有効にするために必要な唯一の変更はfrom functional import seqではなく、 from functional import pseq from functional import seqを使用する場合はpseqを使用することです。 次の操作は、将来のリリースでより多くの機能と並行して実行されます。

  • map / select
  • filter / filter_not / where
  • flat_map

並列化では、複雑な並列処理のPython multiprocessingと縮退チェーンを使用して、オーバーヘッドコストを削減します。 たとえば、一連のマップとフィルタは、 multiprocessingを使用する複数のループではなく、一度にすべて実行されます

ドキュメンテーション

ショートフォームのドキュメントは以下の通りで、完全なドキュメントはdocs.pyfunctional.orgにあります。

ストリームAPI

すべてのPyFunctionalストリームは、 seqオブジェクトを通じてアクセスできます。 ストリームを作成する主な方法は、 seqをiterableで呼び出すことです。 seq callableはスマートで、以下の例に示すように、複数のタイプのパラメータを受け入れることができます。

# Passing a list
seq([1, 1, 2, 3]).to_set()
# [1, 2, 3]

# Passing direct arguments
seq(1, 1, 2, 3).map(lambda x: x).to_list()
# [1, 1, 2, 3]

# Passing a single value
seq(1).map(lambda x: -x).to_list()
# [-1]

seqまた、以下に示すように属性関数として他のストリームへのエントリを提供する。

# number range
seq.range(10)

# text file
seq.open('filepath')

# json file
seq.json('filepath')

# jsonl file
seq.jsonl('filepath')

# csv file
seq.csv('filepath')
seq.csv_dict_reader('filepath')

# sqlite3 db and sql query
seq.sqlite3('filepath', 'select * from data')

これらの関数で使用できるパラメータの詳細については、 ストリームドキュメントを参照してください

変換とアクションAPI

以下は、 seqからストリームオブジェクトに対してseq関数の完全なリストです。 完全なドキュメント参照変換とアクションAPI

関数 説明 タイプ
map(func)/select(func) funcをシーケンスの要素にマップする 変換
starmap(func)/smap(func) itertools.starmapitertools.starmapシーケンスに適用する 変換
filter(func)/where(func) シーケンスの要素をfunc(element)True func(element)のみにフィルタする 変換
filter_not(func) シーケンスの要素をfunc(element)False func(element)のみにフィルタする 変換
flatten() リストのシーケンスを単一のシーケンスにフラット化する 変換
flat_map(func) funcは反復可能性を返さなければなりません。 funcを各要素にマップし、結果を1つのフラットシーケンスにマージします。 変換
group_by(func) グループを(key, value)ペアにグループ化し(key, value)ここで、 key=func(element)valueは元のシーケンス 変換
group_by_key() (key, value)による(key, value)ペアのグループ化 変換
reduce_by_key(func) funcを使って(key, value)ペアのリストをfunc 変換
count_by_key() key出現を(key, value)ペアのリストでカウントし(key, value) 変換
count_by_value() リスト内の各値の出現回数をカウントする 変換
union(other) ユニークな要素の連鎖など 変換
intersection(other) シーケンスやotherユニークな要素の交差 変換
difference(other) ユニークな要素がシーケンス内に存在するが、 otherは存在しない新しいシーケンス 変換
symmetric_difference(other) シーケンスまたはother要素に存在する固有の要素を持つ新しいシーケンスですが、両方ではありません 変換
distinct() シーケンスの異なる要素を返します。 要素はハッシュ可能でなければならない 変換
distinct_by(func) funcをキーとして使用してシーケンスの異なる要素を返す 変換
drop(n) シーケンスの最初のn要素を削除する 変換
drop_right(n) シーケンスの最後のn要素を削除する 変換
drop_while(func) funcTrueに評価されている間に要素をfunc 、残りを返します 変換
take(n) 最初のn要素のシーケンスを返す 変換
take_while(func) funcTrueに評価されている間に要素をfunc 、残りの要素を削除します。 変換
init() 最後の要素がないシーケンスを返す 変換
tail() 最初の要素のないシーケンスを返す 変換
inits() シーケンスの連続したinitsを返します。 変換
tails() シーケンスの連続したテールを返します。 変換
zip(other) シーケンスをotherと一緒にother 変換
zip_with_index(start=0) インデックスのシーケンスを右端のstartします 変換
enumerate(start=0) 左端のstart点から始まるインデックスでシーケンスを圧縮します。 変換
cartesian(*iterables, repeat=1) itertools.productからデカルト積を返します。 変換
inner_join(other) 他のシーケンスとの内部結合を返します。 一連の(key, value)ペアでなければならない 変換
outer_join(other) 他のシーケンスとの外部結合を返します。 一連の(key, value)ペアでなければならない 変換
left_join(other) シーケンスの左結合をotherと返します。 一連の(key, value)ペアでなければならない 変換
right_join(other) シーケンスと他のシーケンスとの右結合を返します。 一連の(key, value)ペアでなければならない 変換
join(other, join_type='inner') join_type指定された他のものとのシーケンスの結合を返します。 一連の(key, value)ペアでなければならない 変換
partition(func) func(element)を満たす要素とそうでない要素にシーケンスを分割する 変換
grouped(size) 要素をサイズsizeグループに分割する 変換
sorted(key=None, reverse=False)/order_by(func) Pythonに従ってソートされた要素をソートして返す 変換
reverse() 逆順を返します。 変換
slice(start, until) 開始時start最大〜 until要素を含むシーケンス 変換
head() / first() 順番に最初の要素を返します。 アクション
head_option() シーケンスの最初の要素を返します。空の場合はNone返します。 アクション
last() 最後の要素を順番に返します。 アクション
last_option() 最後の要素を順番に返します。空の場合はNone返します。 アクション
len() / size() シーケンスの長さを返す アクション
count(func) func(element)がTrueである順に要素の数を返します アクション
empty() シーケンスの長さがゼロの場合はTrue返します。 アクション
non_empty() シーケンスの長さがゼロ以外の場合はTrue返します。 アクション
all() シーケンス内のすべての要素がTrue場合にTrue返します。 アクション
exists(func) シーケンス内の任意の要素のfunc(element) True場合にTrue返します。 アクション
for_all(func) シーケンス内のすべての要素に対してfunc(element)True場合はTrueを返します。 アクション
find(func) func(element)を最初に評価する要素をTrue func(element) True アクション
any() シーケンス内の要素がTrueあればTrue返します。 アクション
max() 順番に最大要素を返します。 アクション
min() 順番に最小要素を返す アクション
max_by(func) 最大値func(element)持つ要素を返します。 アクション
min_by(func) 最小値を持つ要素を返すfunc(element) アクション
sum()/sum(projection) 投影を使用する可能性のある要素の合計を返します。 アクション
product()/product(projection) プロジェクションを使用する可能性のある要素の積を返します。 アクション
average()/average(projection) 投影を使用する可能性のある要素の平均を返します。 アクション
aggregate(func)/aggregate(seed, func)/aggregate(seed, func, result_map) リストのseedまたは最初の要素で始まるfuncを使用して集計し、 result_mapを結果に適用します アクション
fold_left(zero_value, func) funcと初期値zero_valueを使用して、要素を左から右にzero_valueますzero_value アクション
fold_right(zero_value, func) funcと初期値zero_valueを使用して要素を右から左にzero_valueますzero_value アクション
make_string(separator) str(element)間にseparator文字を含む文字列を返します。 アクション
dict(default=None) / to_dict(default=None) (Key, Value)ペアのシーケンスをdictionary変換します。 defaultがNoneでない場合、それはcollections.defaultdict作成に使用される値またはゼロ引き数の呼び出し可能でなければなりません アクション
list() / to_list() シーケンスをリストに変換します。 アクション
set() / to_set() シーケンスをセットに変換します。 アクション
to_file(path) 改行の各要素を含むパスのファイルにシーケンスを保存します。 アクション
to_csv(path) パスにあるcsvファイルにシーケンスを保存します。各エレメントは行を表します。 アクション
to_jsonl(path) シーケンスをjsonlファイルに保存し、各要素はjsonに変換され、新しい行に出力されます アクション
to_json(path) シーケンスをjsonファイルに保存します。 内容は、jsonのルートが配列か辞書かによって決まります アクション
to_sqlite3(conn, tablename_or_query, *args, **kwargs) シーケンスをSQLite3 dbに保存します。 ターゲットテーブルは事前に作成する必要があります。 アクション
to_pandas(columns=None) シーケンスをpandas DataFrameに変換します。 アクション
cache() 直ちにシーケンスの評価を強制し、結果をキャッシュする アクション
for_each(func) シーケンスの各要素に対してfuncを実行します。 アクション

遅延実行

可能な限り、 PyFunctionalは遅れて計算します。 これは、シーケンスに適用された変換のリストを追跡し、アクションが呼び出されたときにのみ評価します。 PyFunctionalこれは追跡系統と呼ばれます。 これは、 PyFunctionalが計算結果をキャッシュして、高価な再計算を防止する機能も担っています。 これは、賢明な行動を維持し、控えめに使用するために主に行われます。 たとえば、 size()を呼び出すと、元のシーケンスがキャッシュされます。 これが行われず、入力がイテレータであった場合、それ以降のコールは、長さの計算に使用されたため、期限切れのイテレータで動作します。 同様に、 reprもキャッシュします。これは、対話型セッションで最も頻繁に使用されるため、同じ値を再計算することが望ましくないためです。 以下は、系統の検査の例です。

def times_2(x):
    print(x)
    return 2 * x
elements = seq(1, 1, 2, 3, 4).map(times_2).distinct()
elements._lineage
# Lineage: sequence -> map(times_2) -> distinct

l_elements = elements.to_list()
# Prints: 1
# Prints: 1
# Prints: 2
# Prints: 3
# Prints: 4

elements._lineage
# Lineage: sequence -> map(times_2) -> distinct -> cache

l_elements = elements.to_list()
# The cached result is returned so times_2 is not called and nothing is printed

seq.openおよび関連するAPIを使用してファイルをseq.openと、ファイルに特別な処理が行われます。 functional.util.ReusableFileは、標準的なpythonファイルの周りにラッパーを実装し、1つのファイルオブジェクトに対して複数の反復をサポートし、反復の終了とファイルの終了を正しく処理します。

ロードマップの考え方

  • SQLベースのクエリプランナーとインタプリタ
  • _ラムダ演算子
  • 1.0次のリリースを準備する

貢献とバグ修正

寄稿やバグ報告は大歓迎です。 これまでのところ、プルリクエストの受け入れ率は100%であり、コントリビュータは貴重なフィードバックとコードに対する批判を提供しています。 パッケージのユーザーから、特にそのパッケージが使用されているもの、うまくいくもの、改善できるものを聞くことは素晴らしいことです。

貢献するには、 PyFunctionalフォークを作成し、変更を加え、 TravisCIで実行していることを確認します(あなたはアカウントにサインアップし、Githubをリンクする必要があります)。 マージするには、すべてのプルリクエストが以下の条件を満たす必要があります。

  • すべての単体テストを渡す
  • すべてのpylintテストに合格するか、またはその理由を説明した警告を無視する
  • coveralls.ioで100%テストカバレッジを達成する)
  • 変更を含むNext Release見出しのCHANGELOG.mdファイルを編集します。

接触

チャットのジッター

サポートされているPythonのバージョン

PyFunctionalサポートは、Python 2.7,3.4.4,3.5,3.6、およびPyPyに対してテストされています

変更ログ

変更ログ

私について

私(著者)の詳細については、 pedrorodriguez.ioの私のウェブページをご覧ください

私はボルダーのコロラド大学でコンピュータサイエンスの博士号を取得しています。 私の研究対象は、大規模機械学習、分散コンピューティング、および隣接分野です。 私は2015年にUC Berkeleyでコンピュータサイエンスの学士号を取得しました。以前はApache SparkでUC Berkeley AMPLabの研究を行い、データ科学者としてTruliaで働いていました。これまでの夏、Oracle Data Cloudのデータ科学者として働いていました。

TruliaでPythonを広範囲に使用しながらPyFunctionalを作成し、Spark RDDやScalaコレクションが持つデータ操作の使いやすさが欠けていたことが分かりました。 このプロジェクトでは、これらのAPIとLINQのベストアイデアを利用して、Scalaを使用するとデータを操作する簡単な方法を提供したり、PySparkが過度に使用されたりすることはありません。

貢献者

これらの人々は、 PyFunctional改善に惜しみなく貢献しました







-EntilZha
-, , , ,

執筆者: