Project

General

Profile

Python SDK » History » Version 7

Brett Smith, 09/23/2014 02:02 PM
fix bugs in first example script

1 1 Tom Clegg
h1. Python SDK
2
3
(design draft)
4
5 5 Brett Smith
h1. Hypothetical future Crunch scripts
6 1 Tom Clegg
7 5 Brett Smith
We're writing these out with the goal of designing a new SDK for Crunch script authors.
8
9
{{toc}}
10
11
h2. Example scripts
12
13 6 Brett Smith
h3. Example: "Normalize" files matching a regexp
14 5 Brett Smith
15 1 Tom Clegg
<pre><code class="python">
16
#!/usr/bin/env python
17
18
from arvados import CrunchJob
19
20
import examplelib
21
import re
22
23
class NormalizeMatchingFiles(CrunchJob):
24
    @CrunchJob.task()
25
    def grep_files(self):
26
        # CrunchJob instantiates input parameters based on the
27
        # dataclass attribute.  When we ask for the input parameter,
28
        # CrunchJob sees that it's a Collection, and returns a
29
        # CollectionReader object.
30 7 Brett Smith
        input_collection = self.job_param('input')
31
        for filename in input_collection.filenames():
32
            self.grep_file(self.job_param('pattern'), input_collection, filename)
33 1 Tom Clegg
34
    @CrunchJob.task()
35 3 Brett Smith
    def grep_file(self, pattern, collection, filename):
36
        regexp = re.compile(pattern)
37
        with collection.open(filename) as in_file:
38 1 Tom Clegg
            for line in in_file:
39
                if regexp.search(line):
40 4 Brett Smith
                    self.normalize(in_file)
41 1 Tom Clegg
                    break
42
43
    # examplelib is already multi-threaded and will peg the whole
44
    # compute node.  These tasks should run sequentially.
45 4 Brett Smith
    # When tasks are created, Arvados-specific objects like Collection file
46
    # objects are serialized as task parameters.  CrunchJob instantiates
47
    # these parameters as real objects when it runs the task.
48 1 Tom Clegg
    @CrunchJob.task(parallel_with=[])
49 7 Brett Smith
    def normalize(self, collection_file):
50
        output = examplelib.frob(collection_file.mount_path())
51 1 Tom Clegg
        # self.output is a CollectionWriter.  When this task method finishes,
52
        # CrunchJob checks if we wrote anything to it.  If so, it takes care
53
        # of finishing the upload process, and sets this task's output to the
54
        # Collection UUID.
55 7 Brett Smith
        with self.output.open(collection_file.name) as out_file:
56 1 Tom Clegg
            out_file.write(output)
57
58
59
if __name__ == '__main__':
60 7 Brett Smith
    NormalizeMatchingFiles(task0='grep_files').run()
61 1 Tom Clegg
</code></pre>
62 7 Brett Smith
63 6 Brett Smith
64
h3. Example: Crunch statistics on a Collection of fastj files indicating a tumor
65
66
This demonstrates scheduling tasks in order and explicit job output.
67
68
<pre><code class="python">
69
#!/usr/bin/env python
70
71
import glob
72
import os
73
import pprint
74
75
from arvados import CrunchJob
76
from subprocess import check_call
77
78
class TumorAnalysis(CrunchJob):
79
    OUT_EXT = '.analysis'
80
81
    @CrunchJob.task()
82
    def check_fastjs(self):
83
        in_coll = self.job_param('input')
84
        for name in in_coll.filenames():
85
            if name.endswith('.fastj'):
86
                self.classify(in_coll, name)
87
        # analyze_tumors gets scheduled to run after all the classification,
88
        # since they're not parallel with each other (and it was invoked later).
89
        self.analyze_tumors()
90
91
    @CrunchJob.task()
92
    def classify(self, collection, filename):
93
        # Methods that refer to directories, like mount_path and job_dir,
94
        # work like os.path.join when you pass them arguments.
95
        check_call(['normal_or_tumor', collection.mount_path(filename)])
96
        outpath = filename + self.OUT_EXT
97
        with open(outpath) as result:
98
             is_tumor = 'tumor' in result.read(4096)
99
        if is_tumor:
100
            os.rename(outpath, self.job_dir(outpath))
101
102
    @CrunchJob.task()
103
    def analyze_tumors(self):
104
        compiled = {}
105
        results = glob.glob(self.job_dir('*' + self.OUT_EXT))
106
        for outpath in results:
107
            with open(outpath) as outfile:
108
                compiled[thing] = ...  # Imagine this is a reduce-type step.
109
        # job_output is a CollectionWriter.  Writing to it overrides the
110
        # default behavior where job output is collated task output.
111
        with self.job_output.open('compiled_numbers.log') as resultfile:
112
            pprint.pprint(compiled, resultfile)
113
114
115
if __name__ == '__main__':
116
    TumorAnalysis(task0='check_fastjs').run()
117 1 Tom Clegg
</code></pre>
118
119 5 Brett Smith
h3. Example from #3603
120
121
This is the script that Abram used to illustrate #3603.
122
123
<pre><code class="python">
124
#!/usr/bin/env python
125
126
from arvados import Collection, CrunchJob
127
from subprocess import check_call
128
129
class Example3603(CrunchJob):
130
    @CrunchJob.task()
131
    def parse_human_map(self):
132
        refpath = self.job_param('REFPATH').name
133
        for line in self.job_param('HUMAN_COLLECTION_LIST'):
134
            fastj_id, human_id = line.strip().split(',')
135
            self.run_ruler(refpath, fastj_id, human_id)
136
137
    @CrunchJob.task()
138
    def run_ruler(self, refpath, fastj_id, human_id):
139
        check_call(["tileruler", "--crunch", "--noterm", "abv",
140
                    "-human", human_id,
141
                    "-fastj-path", Collection(fastj_id).mount_path(),
142
                    "-lib-path", refpath])
143
        self.output.add('.')  # Or the path where tileruler writes output.
144
145
146
if __name__ == '__main__':
147
    Example3603(task0='parse_human_map').run()
148
</code></pre>
149
150
h2. Notes/TODO
151
152 2 Tom Clegg
* Important concurrency limits that job scripts must be able to express:
153
** Task Z cannot start until all outputs/side effects of tasks W, X, Y are known/complete (e.g., because Z uses WXY's outputs as its inputs).
154
** Task Y and Z cannot run on the same worker node without interfering with each other (e.g., due to RAM requirements).
155
* In general, the output name is not known until the task is nearly finished. Frequently it is clearer to specify it when the task is queued, though. We should provide a convenient way to do this without any boilerplate in the queued task.
156
* A second example that uses a "case" and "control" input (e.g., "tumor" and "normal") might help reveal features.
157 1 Tom Clegg
* Should get more clear about how the output of the job (as opposed to the output of the last task) is to be set. The obvious way (concatenate all task outputs) should be a one-liner, if not implicit. Either way, it should run in a task rather than being left up to @crunch-job@.