This repository has been archived by the owner on Jun 5, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 1
/
main.go
78 lines (64 loc) · 1.62 KB
/
main.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
package mapreduce
import (
"bytes"
"flag"
"fmt"
"os"
)
// InsideJob determines whether current process is running inside a mapreduce job.
func InsideJob() bool {
return os.Getenv("YT_JOB_ID") != ""
}
type jobArgs struct {
job string
nOutputPipes int
}
func parseJobArgs(args []string) (job jobArgs, err error) {
flags := flag.NewFlagSet("job", flag.ExitOnError)
flags.StringVar(&job.job, "job", "", "")
flags.IntVar(&job.nOutputPipes, "output-pipes", 0, "")
err = flags.Parse(args)
return
}
// JobMain runs user code inside mapreduce job.
//
// Binary that wishes to run mapreduce operations must place the following code
// at the beginning of the main() function.
//
// if mapreduce.InsideJob() {
// os.Exit(mapreduce.JobMain())
// }
func JobMain() int {
args, err := parseJobArgs(os.Args[1:])
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
}
var ctx jobContext
if err = ctx.initEnv(); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
return 5
}
stateBlob, err := readJobState(&ctx)
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
return 6
}
state, err := decodeJob(bytes.NewBuffer(stateBlob))
if err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
return 7
}
if err = ctx.initPipes(state, args.nOutputPipes); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
return 3
}
if err := state.Job.Do(&ctx, ctx.in, ctx.writers()); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
return 1
}
if err := ctx.finish(); err != nil {
_, _ = fmt.Fprintf(os.Stderr, "job: %+v\n", err)
return 4
}
return 0
}